Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-16229 Cleaning up size and heapSize calculation.

This commit is contained in:
anoopsamjohn 2016-09-13 11:43:26 +05:30 committed by Gary Helmling
parent 2566cfeb60
commit 77b327320a
17 changed files with 419 additions and 385 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
hbase-common/src/main/java/org/apache/hadoop/hbase/util/ClassSize.java
View File

@ -46,6 +46,12 @@ public class ClassSize {
/** Overhead for ArrayList(0) */
public static final int ARRAYLIST;
/** Overhead for LinkedList(0) */
public static final int LINKEDLIST;
/** Overhead for a single entry in LinkedList */
public static final int LINKEDLIST_ENTRY;
/** Overhead for ByteBuffer */
public static final int BYTE_BUFFER;
@ -100,6 +106,9 @@ public class ClassSize {
/** Overhead for AtomicBoolean */
public static final int ATOMIC_BOOLEAN;
/** Overhead for AtomicReference */
public static final int ATOMIC_REFERENCE;
/** Overhead for CopyOnWriteArraySet */
public static final int COPYONWRITE_ARRAYSET;
@ -240,6 +249,10 @@ public class ClassSize {
ARRAYLIST = align(OBJECT + REFERENCE + (2 * Bytes.SIZEOF_INT)) + align(ARRAY);
LINKEDLIST = align(OBJECT + (2 * Bytes.SIZEOF_INT) + (2 * REFERENCE));
LINKEDLIST_ENTRY = align(OBJECT + (2 * REFERENCE));
//noinspection PointlessArithmeticExpression
BYTE_BUFFER = align(OBJECT + REFERENCE +
(5 * Bytes.SIZEOF_INT) +
@ -292,6 +305,8 @@ public class ClassSize {
ATOMIC_BOOLEAN = align(OBJECT + Bytes.SIZEOF_BOOLEAN);
ATOMIC_REFERENCE = align(OBJECT + REFERENCE);
COPYONWRITE_ARRAYSET = align(OBJECT + REFERENCE);
COPYONWRITE_ARRAYLIST = align(OBJECT + (2 * REFERENCE) + ARRAY);

49
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java
View File

@ -52,34 +52,29 @@ public abstract class AbstractMemStore implements MemStore {
private final CellComparator comparator;
// active segment absorbs write operations
private volatile MutableSegment active;
protected volatile MutableSegment active;
// Snapshot of memstore. Made for flusher.
private volatile ImmutableSegment snapshot;
protected volatile ImmutableSegment snapshot;
protected volatile long snapshotId;
// Used to track when to flush
private volatile long timeOfOldestEdit;
public final static long FIXED_OVERHEAD = ClassSize.align(
ClassSize.OBJECT +
(4 * ClassSize.REFERENCE) +
(2 * Bytes.SIZEOF_LONG));
public final static long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD +
(ClassSize.ATOMIC_LONG + ClassSize.TIMERANGE_TRACKER +
ClassSize.CELL_SET + ClassSize.CONCURRENT_SKIPLISTMAP));
public final static long FIXED_OVERHEAD = ClassSize
.align(ClassSize.OBJECT + (4 * ClassSize.REFERENCE) + (2 * Bytes.SIZEOF_LONG));
public final static long DEEP_OVERHEAD = FIXED_OVERHEAD;
protected AbstractMemStore(final Configuration conf, final CellComparator c) {
this.conf = conf;
this.comparator = c;
resetActive();
this.snapshot = SegmentFactory.instance().createImmutableSegment(c, 0);
this.snapshot = SegmentFactory.instance().createImmutableSegment(c);
this.snapshotId = NO_SNAPSHOT_ID;
}
protected void resetActive() {
// Reset heap to not include any keys
this.active = SegmentFactory.instance().createMutableSegment(conf, comparator, DEEP_OVERHEAD);
this.active = SegmentFactory.instance().createMutableSegment(conf, comparator);
this.timeOfOldestEdit = Long.MAX_VALUE;
}
@ -200,8 +195,7 @@ public abstract class AbstractMemStore implements MemStore {
// create a new snapshot and let the old one go.
Segment oldSnapshot = this.snapshot;
if (!this.snapshot.isEmpty()) {
this.snapshot = SegmentFactory.instance().createImmutableSegment(
getComparator(), 0);
this.snapshot = SegmentFactory.instance().createImmutableSegment(this.comparator);
}
this.snapshotId = NO_SNAPSHOT_ID;
oldSnapshot.close();
@ -213,12 +207,12 @@ public abstract class AbstractMemStore implements MemStore {
*/
@Override
public long heapSize() {
return getActive().getSize();
return size();
}
@Override
public long getSnapshotSize() {
return getSnapshot().getSize();
return this.snapshot.keySize();
}
@Override
@ -385,7 +379,7 @@ public abstract class AbstractMemStore implements MemStore {
// so we cant add the new Cell w/o knowing what's there already, but we also
// want to take this chance to delete some cells. So two loops (sad)
SortedSet<Cell> ss = getActive().tailSet(firstCell);
SortedSet<Cell> ss = this.active.tailSet(firstCell);
for (Cell cell : ss) {
// if this isnt the row we are interested in, then bail:
if (!CellUtil.matchingColumn(cell, family, qualifier)
@ -433,38 +427,33 @@ public abstract class AbstractMemStore implements MemStore {
}
}
/**
* @return The size of the active segment. Means sum of all cell's size.
*/
protected long keySize() {
return heapSize() - DEEP_OVERHEAD;
return this.active.keySize();
}
protected CellComparator getComparator() {
return comparator;
}
protected MutableSegment getActive() {
@VisibleForTesting
MutableSegment getActive() {
return active;
}
protected ImmutableSegment getSnapshot() {
@VisibleForTesting
ImmutableSegment getSnapshot() {
return snapshot;
}
protected AbstractMemStore setSnapshot(ImmutableSegment snapshot) {
this.snapshot = snapshot;
return this;
}
protected void setSnapshotSize(long snapshotSize) {
getSnapshot().setSize(snapshotSize);
}
/**
* Check whether anything need to be done based on the current active set size
*/
protected abstract void checkActiveSize();
/**
* Returns an ordered list of segments from most recent to oldest in memstore
* @return an ordered list of segments from most recent to oldest in memstore
*/
protected abstract List<Segment> getSegments() throws IOException;

92
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java
View File

@ -22,7 +22,6 @@ import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
@ -52,13 +51,6 @@ import org.apache.hadoop.hbase.wal.WAL;
@InterfaceAudience.Private
public class CompactingMemStore extends AbstractMemStore {
public final static long DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM = ClassSize.align(
ClassSize.TIMERANGE_TRACKER + ClassSize.CELL_SET + ClassSize.CONCURRENT_SKIPLISTMAP);
public final static long DEEP_OVERHEAD_PER_PIPELINE_CELL_ARRAY_ITEM = ClassSize.align(
ClassSize.TIMERANGE_TRACKER + ClassSize.TIMERANGE +
ClassSize.CELL_SET + ClassSize.CELL_ARRAY_MAP);
// Default fraction of in-memory-flush size w.r.t. flush-to-disk size
public static final String IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY =
"hbase.memstore.inmemoryflush.threshold.factor";
@ -75,6 +67,13 @@ public class CompactingMemStore extends AbstractMemStore {
@VisibleForTesting
private final AtomicBoolean allowCompaction = new AtomicBoolean(true);
public static final long DEEP_OVERHEAD = AbstractMemStore.DEEP_OVERHEAD
+ 6 * ClassSize.REFERENCE // Store, RegionServicesForStores, CompactionPipeline,
// MemStoreCompactor, inMemoryFlushInProgress, allowCompaction
+ Bytes.SIZEOF_LONG // inmemoryFlushSize
+ 2 * ClassSize.ATOMIC_BOOLEAN// inMemoryFlushInProgress and allowCompaction
+ CompactionPipeline.DEEP_OVERHEAD + MemStoreCompactor.DEEP_OVERHEAD;
public CompactingMemStore(Configuration conf, CellComparator c,
HStore store, RegionServicesForStores regionServices) throws IOException {
super(conf, c);
@ -100,28 +99,18 @@ public class CompactingMemStore extends AbstractMemStore {
LOG.info("Setting in-memory flush size threshold to " + inmemoryFlushSize);
}
public static long getSegmentSize(Segment segment) {
return segment.keySize();
}
public static long getSegmentsSize(List<? extends Segment> list) {
long res = 0;
for (Segment segment : list) {
res += getSegmentSize(segment);
}
return res;
}
/**
* @return Total memory occupied by this MemStore.
* This is not thread safe and the memstore may be changed while computing its size.
* It is the responsibility of the caller to make sure this doesn't happen.
* @return Total memory occupied by this MemStore. This includes active segment size and heap size
* overhead of this memstore but won't include any size occupied by the snapshot. We
* assume the snapshot will get cleared soon. This is not thread safe and the memstore may
* be changed while computing its size. It is the responsibility of the caller to make
* sure this doesn't happen.
*/
@Override
public long size() {
long res = 0;
for (Segment item : getSegments()) {
res += item.getSize();
long res = DEEP_OVERHEAD + this.active.size();
for (Segment item : pipeline.getSegments()) {
res += CompactionPipeline.ENTRY_OVERHEAD + item.size();
}
return res;
}
@ -131,11 +120,13 @@ public class CompactingMemStore extends AbstractMemStore {
* The store may do any post-flush actions at this point.
* One example is to update the WAL with sequence number that is known only at the store level.
*/
@Override public void finalizeFlush() {
@Override
public void finalizeFlush() {
updateLowestUnflushedSequenceIdInWAL(false);
}
@Override public boolean isSloppy() {
@Override
public boolean isSloppy() {
return true;
}
@ -148,10 +139,9 @@ public class CompactingMemStore extends AbstractMemStore {
*/
@Override
public MemStoreSnapshot snapshot() {
MutableSegment active = getActive();
// If snapshot currently has entries, then flusher failed or didn't call
// cleanup. Log a warning.
if (!getSnapshot().isEmpty()) {
if (!this.snapshot.isEmpty()) {
LOG.warn("Snapshot called again without clearing previous. " +
"Doing nothing. Another ongoing flush or did we fail last attempt?");
} else {
@ -161,21 +151,22 @@ public class CompactingMemStore extends AbstractMemStore {
+ getFamilyName());
}
stopCompaction();
pushActiveToPipeline(active);
pushActiveToPipeline(this.active);
snapshotId = EnvironmentEdgeManager.currentTime();
pushTailToSnapshot();
}
return new MemStoreSnapshot(snapshotId, getSnapshot());
return new MemStoreSnapshot(snapshotId, this.snapshot);
}
/**
* On flush, how much memory we will clear.
* @return size of data that is going to be flushed
*/
@Override public long getFlushableSize() {
long snapshotSize = getSnapshot().getSize();
if(snapshotSize == 0) {
//if snapshot is empty the tail of the pipeline is flushed
@Override
public long getFlushableSize() {
long snapshotSize = getSnapshotSize();
if (snapshotSize == 0) {
// if snapshot is empty the tail of the pipeline is flushed
snapshotSize = pipeline.getTailSize();
}
return snapshotSize > 0 ? snapshotSize : keySize();
@ -186,7 +177,7 @@ public class CompactingMemStore extends AbstractMemStore {
long minSequenceId = pipeline.getMinSequenceId();
if(minSequenceId != Long.MAX_VALUE) {
byte[] encodedRegionName = getRegionServices().getRegionInfo().getEncodedNameAsBytes();
byte[] familyName = getFamilyNameInByte();
byte[] familyName = getFamilyNameInBytes();
WAL WAL = getRegionServices().getWAL();
if (WAL != null) {
WAL.updateStore(encodedRegionName, familyName, minSequenceId, onlyIfGreater);
@ -197,10 +188,10 @@ public class CompactingMemStore extends AbstractMemStore {
@Override
public List<Segment> getSegments() {
List<Segment> pipelineList = pipeline.getSegments();
List<Segment> list = new LinkedList<Segment>();
list.add(getActive());
List<Segment> list = new ArrayList<Segment>(pipelineList.size() + 2);
list.add(this.active);
list.addAll(pipelineList);
list.add(getSnapshot());
list.add(this.snapshot);
return list;
}
@ -235,7 +226,7 @@ public class CompactingMemStore extends AbstractMemStore {
}
public String getFamilyName() {
return Bytes.toString(getFamilyNameInByte());
return Bytes.toString(getFamilyNameInBytes());
}
@Override
@ -248,12 +239,12 @@ public class CompactingMemStore extends AbstractMemStore {
// The list of elements in pipeline + the active element + the snapshot segment
// TODO : This will change when the snapshot is made of more than one element
List<KeyValueScanner> list = new ArrayList<KeyValueScanner>(pipelineList.size() + 2);
list.add(getActive().getScanner(readPt, order + 1));
list.add(this.active.getScanner(readPt, order + 1));
for (Segment item : pipelineList) {
list.add(item.getScanner(readPt, order));
order--;
}
list.add(getSnapshot().getScanner(readPt, order));
list.add(this.snapshot.getScanner(readPt, order));
return Collections.<KeyValueScanner> singletonList(new MemStoreScanner(getComparator(), list));
}
@ -291,11 +282,10 @@ public class CompactingMemStore extends AbstractMemStore {
// Phase I: Update the pipeline
getRegionServices().blockUpdates();
try {
MutableSegment active = getActive();
if (LOG.isDebugEnabled()) {
LOG.debug("IN-MEMORY FLUSH: Pushing active segment into compaction pipeline");
}
pushActiveToPipeline(active);
pushActiveToPipeline(this.active);
} finally {
getRegionServices().unblockUpdates();
}
@ -319,7 +309,7 @@ public class CompactingMemStore extends AbstractMemStore {
}
}
private byte[] getFamilyNameInByte() {
private byte[] getFamilyNameInBytes() {
return store.getFamily().getName();
}
@ -328,7 +318,7 @@ public class CompactingMemStore extends AbstractMemStore {
}
private boolean shouldFlushInMemory() {
if(getActive().getSize() > inmemoryFlushSize) { // size above flush threshold
if (this.active.size() > inmemoryFlushSize) { // size above flush threshold
// the inMemoryFlushInProgress is CASed to be true here in order to mutual exclude
// the insert of the active into the compaction pipeline
return (inMemoryFlushInProgress.compareAndSet(false,true));
@ -350,8 +340,6 @@ public class CompactingMemStore extends AbstractMemStore {
private void pushActiveToPipeline(MutableSegment active) {
if (!active.isEmpty()) {
long delta = DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM - DEEP_OVERHEAD;
active.incSize(delta);
pipeline.pushHead(active);
resetActive();
}
@ -360,9 +348,7 @@ public class CompactingMemStore extends AbstractMemStore {
private void pushTailToSnapshot() {
ImmutableSegment tail = pipeline.pullTail();
if (!tail.isEmpty()) {
setSnapshot(tail);
long size = getSegmentSize(tail);
setSnapshotSize(size);
this.snapshot = tail;
}
}
@ -428,7 +414,7 @@ public class CompactingMemStore extends AbstractMemStore {
// debug method
public void debug() {
String msg = "active size="+getActive().getSize();
String msg = "active size=" + this.active.size();
msg += " threshold="+IN_MEMORY_FLUSH_THRESHOLD_FACTOR_DEFAULT* inmemoryFlushSize;
msg += " allow compaction is "+ (allowCompaction.get() ? "true" : "false");
msg += " inMemoryFlushInProgress is "+ (inMemoryFlushInProgress.get() ? "true" : "false");

31
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java
View File

@ -24,7 +24,10 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
/**
* The compaction pipeline of a {@link CompactingMemStore}, is a FIFO queue of segments.
@ -39,13 +42,17 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience;
public class CompactionPipeline {
private static final Log LOG = LogFactory.getLog(CompactionPipeline.class);
public final static long FIXED_OVERHEAD = ClassSize
.align(ClassSize.OBJECT + (2 * ClassSize.REFERENCE) + Bytes.SIZEOF_LONG);
public final static long DEEP_OVERHEAD = FIXED_OVERHEAD + ClassSize.LINKEDLIST;
public final static long ENTRY_OVERHEAD = ClassSize.LINKEDLIST_ENTRY;
private final RegionServicesForStores region;
private LinkedList<ImmutableSegment> pipeline;
private long version;
private static final ImmutableSegment EMPTY_MEM_STORE_SEGMENT = SegmentFactory.instance()
.createImmutableSegment(null,
CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM);
.createImmutableSegment((CellComparator) null);
public CompactionPipeline(RegionServicesForStores region) {
this.region = region;
@ -105,8 +112,8 @@ public class CompactionPipeline {
}
if (region != null) {
// update the global memstore size counter
long suffixSize = CompactingMemStore.getSegmentsSize(suffix);
long newSize = CompactingMemStore.getSegmentSize(segment);
long suffixSize = getSegmentsKeySize(suffix);
long newSize = segment.keySize();
long delta = suffixSize - newSize;
long globalMemstoreSize = region.addAndGetGlobalMemstoreSize(-delta);
if (LOG.isDebugEnabled()) {
@ -117,6 +124,14 @@ public class CompactionPipeline {
return true;
}
private static long getSegmentsKeySize(List<? extends Segment> list) {
long res = 0;
for (Segment segment : list) {
res += segment.keySize();
}
return res;
}
/**
* If the caller holds the current version, go over the the pipeline and try to flatten each
* segment. Flattening is replacing the ConcurrentSkipListMap based CellSet to CellArrayMap based.
@ -178,20 +193,20 @@ public class CompactionPipeline {
public long getMinSequenceId() {
long minSequenceId = Long.MAX_VALUE;
if(!isEmpty()) {
if (!isEmpty()) {
minSequenceId = pipeline.getLast().getMinSequenceId();
}
return minSequenceId;
}
public long getTailSize() {
if(isEmpty()) return 0;
return CompactingMemStore.getSegmentSize(pipeline.peekLast());
if (isEmpty()) return 0;
return pipeline.peekLast().keySize();
}
private void swapSuffix(LinkedList<ImmutableSegment> suffix, ImmutableSegment segment) {
version++;
for(Segment itemInSuffix : suffix) {
for (Segment itemInSuffix : suffix) {
itemInSuffix.close();
}
pipeline.removeAll(suffix);

30
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
View File

@ -83,20 +83,19 @@ public class DefaultMemStore extends AbstractMemStore {
public MemStoreSnapshot snapshot() {
// If snapshot currently has entries, then flusher failed or didn't call
// cleanup. Log a warning.
if (!getSnapshot().isEmpty()) {
if (!this.snapshot.isEmpty()) {
LOG.warn("Snapshot called again without clearing previous. " +
"Doing nothing. Another ongoing flush or did we fail last attempt?");
} else {
this.snapshotId = EnvironmentEdgeManager.currentTime();
if (!getActive().isEmpty()) {
if (!this.active.isEmpty()) {
ImmutableSegment immutableSegment = SegmentFactory.instance().
createImmutableSegment(getActive());
setSnapshot(immutableSegment);
setSnapshotSize(keySize());
createImmutableSegment(this.active);
this.snapshot = immutableSegment;
resetActive();
}
}
return new MemStoreSnapshot(this.snapshotId, getSnapshot());
return new MemStoreSnapshot(this.snapshotId, this.snapshot);
}
/**
@ -106,7 +105,7 @@ public class DefaultMemStore extends AbstractMemStore {
*/
@Override
public long getFlushableSize() {
long snapshotSize = getSnapshot().getSize();
long snapshotSize = getSnapshotSize();
return snapshotSize > 0 ? snapshotSize : keySize();
}
@ -116,8 +115,8 @@ public class DefaultMemStore extends AbstractMemStore {
*/
public List<KeyValueScanner> getScanners(long readPt) throws IOException {
List<KeyValueScanner> list = new ArrayList<KeyValueScanner>(2);
list.add(getActive().getScanner(readPt, 1));
list.add(getSnapshot().getScanner(readPt, 0));
list.add(this.active.getScanner(readPt, 1));
list.add(this.snapshot.getScanner(readPt, 0));
return Collections.<KeyValueScanner> singletonList(
new MemStoreScanner(getComparator(), list));
}
@ -125,8 +124,8 @@ public class DefaultMemStore extends AbstractMemStore {
@Override
protected List<Segment> getSegments() throws IOException {
List<Segment> list = new ArrayList<Segment>(2);
list.add(getActive());
list.add(getSnapshot());
list.add(this.active);
list.add(this.snapshot);
return list;
}
@ -137,19 +136,16 @@ public class DefaultMemStore extends AbstractMemStore {
*/
Cell getNextRow(final Cell cell) {
return getLowest(
getNextRow(cell, getActive().getCellSet()),
getNextRow(cell, getSnapshot().getCellSet()));
getNextRow(cell, this.active.getCellSet()),
getNextRow(cell, this.snapshot.getCellSet()));
}
@Override public void updateLowestUnflushedSequenceIdInWAL(boolean onlyIfMoreRecent) {
}
/**
* @return Total memory occupied by this MemStore.
*/
@Override
public long size() {
return heapSize();
return this.active.size() + DEEP_OVERHEAD;
}
/**

84
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableSegment.java
View File

@ -21,6 +21,7 @@ package org.apache.hadoop.hbase.regionserver;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.util.ClassSize;
import org.apache.hadoop.hbase.client.Scan;
@ -37,11 +38,12 @@ import java.io.IOException;
*/
@InterfaceAudience.Private
public class ImmutableSegment extends Segment {
/**
* This is an immutable segment so use the read-only TimeRange rather than the heavy-weight
* TimeRangeTracker with all its synchronization when doing time range stuff.
*/
private final TimeRange timeRange;
private static final long DEEP_OVERHEAD = Segment.DEEP_OVERHEAD
+ (2 * ClassSize.REFERENCE) // Refs to timeRange and type
+ ClassSize.TIMERANGE;
public static final long DEEP_OVERHEAD_CSLM = DEEP_OVERHEAD + ClassSize.CONCURRENT_SKIPLISTMAP;
public static final long DEEP_OVERHEAD_CAM = DEEP_OVERHEAD + ClassSize.CELL_ARRAY_MAP;
/**
* Types of ImmutableSegment
@ -51,6 +53,12 @@ public class ImmutableSegment extends Segment {
ARRAY_MAP_BASED,
}
/**
* This is an immutable segment so use the read-only TimeRange rather than the heavy-weight
* TimeRangeTracker with all its synchronization when doing time range stuff.
*/
private final TimeRange timeRange;
private Type type = Type.SKIPLIST_MAP_BASED;
// whether it is based on CellFlatMap or ConcurrentSkipListMap
@ -66,9 +74,8 @@ public class ImmutableSegment extends Segment {
*/
protected ImmutableSegment(Segment segment) {
super(segment);
type = Type.SKIPLIST_MAP_BASED;
TimeRangeTracker trt = getTimeRangeTracker();
this.timeRange = trt == null? null: trt.toTimeRange();
this.type = Type.SKIPLIST_MAP_BASED;
this.timeRange = this.timeRangeTracker == null ? null : this.timeRangeTracker.toTimeRange();
}
/**------------------------------------------------------------------------
@ -80,20 +87,14 @@ public class ImmutableSegment extends Segment {
*/
protected ImmutableSegment(CellComparator comparator, MemStoreCompactorIterator iterator,
MemStoreLAB memStoreLAB, int numOfCells, Type type) {
super(null, // initiailize the CellSet with NULL
comparator, memStoreLAB,
// initial size of segment metadata (the data per cell is added in createCellArrayMapSet)
CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_CELL_ARRAY_ITEM,
ClassSize.CELL_ARRAY_MAP_ENTRY);
super(null, // initiailize the CellSet with NULL
comparator, memStoreLAB);
this.type = type;
// build the true CellSet based on CellArrayMap
CellSet cs = createCellArrayMapSet(numOfCells, iterator);
this.setCellSet(null, cs); // update the CellSet of the new Segment
this.type = type;
TimeRangeTracker trt = getTimeRangeTracker();
this.timeRange = trt == null? null: trt.toTimeRange();
this.timeRange = this.timeRangeTracker == null ? null : this.timeRangeTracker.toTimeRange();
}
/**------------------------------------------------------------------------
@ -101,15 +102,11 @@ 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, MemStoreLAB memStoreLAB) {
super(new CellSet(comparator), // initiailize the CellSet with empty CellSet
comparator, memStoreLAB,
// initial size of segment metadata (the data per cell is added in internalAdd)
CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM,
ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY);
protected ImmutableSegment(CellComparator comparator, MemStoreCompactorIterator iterator,
MemStoreLAB memStoreLAB) {
super(new CellSet(comparator), // initiailize the CellSet with empty CellSet
comparator, memStoreLAB);
type = Type.SKIPLIST_MAP_BASED;
while (iterator.hasNext()) {
Cell c = iterator.next();
// The scanner is doing all the elimination logic
@ -118,9 +115,7 @@ public class ImmutableSegment extends Segment {
boolean usedMSLAB = (newKV != c);
internalAdd(newKV, usedMSLAB); //
}
type = Type.SKIPLIST_MAP_BASED;
TimeRangeTracker trt = getTimeRangeTracker();
this.timeRange = trt == null? null: trt.toTimeRange();
this.timeRange = this.timeRangeTracker == null ? null : this.timeRangeTracker.toTimeRange();
}
///////////////////// PUBLIC METHODS /////////////////////
@ -144,14 +139,16 @@ public class ImmutableSegment extends Segment {
return this.timeRange.getMin();
}
@Override
public long keySize() {
switch (type){
public long size() {
switch (this.type) {
case SKIPLIST_MAP_BASED:
return size.get() - CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM;
return keySize() + DEEP_OVERHEAD_CSLM;
case ARRAY_MAP_BASED:
return size.get() - CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_CELL_ARRAY_ITEM;
default: throw new IllegalStateException();
return keySize() + DEEP_OVERHEAD_CAM;
default:
throw new RuntimeException("Unknown type " + type);
}
}
@ -171,9 +168,6 @@ public class ImmutableSegment extends Segment {
CellSet oldCellSet = getCellSet();
int numOfCells = getCellsCount();
// each Cell is now represented in CellArrayMap
constantCellMetaDataSize = ClassSize.CELL_ARRAY_MAP_ENTRY;
// build the new (CellSet CellArrayMap based)
CellSet newCellSet = recreateCellArrayMapSet(numOfCells);
type = Type.ARRAY_MAP_BASED;
@ -214,6 +208,19 @@ public class ImmutableSegment extends Segment {
return new CellSet(cam);
}
protected long heapSizeChange(Cell cell, boolean succ) {
if (succ) {
switch (this.type) {
case SKIPLIST_MAP_BASED:
return ClassSize
.align(ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY + CellUtil.estimatedHeapSizeOf(cell));
case ARRAY_MAP_BASED:
return ClassSize.align(ClassSize.CELL_ARRAY_MAP_ENTRY + CellUtil.estimatedHeapSizeOf(cell));
}
}
return 0;
}
/*------------------------------------------------------------------------*/
// Create CellSet based on CellArrayMap from current ConcurrentSkipListMap based CellSet
// (without compacting iterator)
@ -239,5 +246,4 @@ public class ImmutableSegment extends Segment {
CellArrayMap cam = new CellArrayMap(getComparator(), cells, 0, idx, false);
return new CellSet(cam);
}
}

6
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
View File

@ -126,7 +126,11 @@ public interface MemStore extends HeapSize {
List<KeyValueScanner> getScanners(long readPt) throws IOException;
/**
* @return Total memory occupied by this MemStore.
* @return Total memory occupied by this MemStore. This includes active segment size and heap size
* overhead of this memstore but won't include any size occupied by the snapshot. We
* assume the snapshot will get cleared soon. This is not thread safe and the memstore may
* be changed while computing its size. It is the responsibility of the caller to make
* sure this doesn't happen.
*/
long size();

24
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactor.java
View File

@ -22,6 +22,8 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
@ -38,7 +40,11 @@ import java.util.concurrent.atomic.AtomicBoolean;
* therefore no special synchronization is required.
*/
@InterfaceAudience.Private
class MemStoreCompactor {
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";
@ -59,6 +65,15 @@ class MemStoreCompactor {
static final boolean MEMSTORE_COMPACTOR_AVOID_SPECULATIVE_SCAN_DEFAULT = false;
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
@ -73,13 +88,6 @@ class MemStoreCompactor {
double fraction = 0.8;
int immutCellsNum = 0; // number of immutable for compaction cells
/**
* Types of Compaction
*/
private enum Type {
COMPACT_TO_SKIPLIST_MAP,
COMPACT_TO_ARRAY_MAP
}
private Type type = Type.COMPACT_TO_ARRAY_MAP;

2
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSnapshot.java
View File

@ -36,7 +36,7 @@ public class MemStoreSnapshot {
public MemStoreSnapshot(long id, ImmutableSegment snapshot) {
this.id = id;
this.cellsCount = snapshot.getCellsCount();
this.size = snapshot.getSize();
this.size = snapshot.keySize();
this.timeRangeTracker = snapshot.getTimeRangeTracker();
this.scanner = snapshot.getKeyValueScanner();
this.tagsPresent = snapshot.isTagsPresent();

25
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MutableSegment.java
View File

@ -21,17 +21,21 @@ 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 org.apache.hadoop.hbase.util.ClassSize;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.ClassSize;
import com.google.common.annotations.VisibleForTesting;
/**
* A mutable segment in memstore, specifically the active segment.
*/
@InterfaceAudience.Private
public class MutableSegment extends Segment {
protected MutableSegment(CellSet cellSet, CellComparator comparator, MemStoreLAB memStoreLAB,
long size) {
super(cellSet, comparator, memStoreLAB, size, ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY);
public final static long DEEP_OVERHEAD = Segment.DEEP_OVERHEAD + ClassSize.CONCURRENT_SKIPLISTMAP;
protected MutableSegment(CellSet cellSet, CellComparator comparator, MemStoreLAB memStoreLAB) {
super(cellSet, comparator, memStoreLAB);
}
/**
@ -44,29 +48,28 @@ public class MutableSegment extends Segment {
return internalAdd(cell, mslabUsed);
}
//methods for test
/**
* Returns the first cell in the segment
* @return the first cell in the segment
*/
@VisibleForTesting
Cell first() {
return this.getCellSet().first();
}
@Override
public boolean shouldSeek(Scan scan, long oldestUnexpiredTS) {
return (getTimeRangeTracker().includesTimeRange(scan.getTimeRange())
&& (getTimeRangeTracker().getMax() >= oldestUnexpiredTS));
return (this.timeRangeTracker.includesTimeRange(scan.getTimeRange())
&& (this.timeRangeTracker.getMax() >= oldestUnexpiredTS));
}
@Override
public long getMinTimestamp() {
return getTimeRangeTracker().getMin();
return this.timeRangeTracker.getMin();
}
@Override
public long keySize() {
return size.get() - CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM;
public long size() {
return keySize() + DEEP_OVERHEAD;
}
}

103
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java
View File

@ -24,7 +24,6 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.CellUtil;
@ -33,6 +32,7 @@ import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.ByteRange;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
import com.google.common.annotations.VisibleForTesting;
@ -47,28 +47,31 @@ import com.google.common.annotations.VisibleForTesting;
@InterfaceAudience.Private
public abstract class Segment {
private static final Log LOG = LogFactory.getLog(Segment.class);
final static long FIXED_OVERHEAD = ClassSize.align(ClassSize.OBJECT
+ 5 * ClassSize.REFERENCE // cellSet, comparator, memStoreLAB, size, timeRangeTracker
+ Bytes.SIZEOF_LONG // minSequenceId
+ Bytes.SIZEOF_BOOLEAN); // tagsPresent
public final static long DEEP_OVERHEAD = FIXED_OVERHEAD + ClassSize.ATOMIC_REFERENCE
+ ClassSize.CELL_SET + ClassSize.ATOMIC_LONG + ClassSize.TIMERANGE_TRACKER;
private AtomicReference<CellSet> cellSet= new AtomicReference<CellSet>();
private final CellComparator comparator;
private long minSequenceId;
private volatile MemStoreLAB memStoreLAB;
/* The size includes everything allocated for this segment,
* use keySize() to get only size of the cells */
private MemStoreLAB memStoreLAB;
// Sum of sizes of all Cells added to this Segment. Cell's heapSize is considered. This is not
// including the heap overhead of this class.
protected final AtomicLong size;
protected final TimeRangeTracker timeRangeTracker;
protected volatile boolean tagsPresent;
private final TimeRangeTracker timeRangeTracker;
protected long constantCellMetaDataSize;
protected Segment(
CellSet cellSet, CellComparator comparator, MemStoreLAB memStoreLAB, long size,
long constantCellSize) {
// This constructor is used to create empty Segments.
protected Segment(CellSet cellSet, CellComparator comparator, MemStoreLAB memStoreLAB) {
this.cellSet.set(cellSet);
this.comparator = comparator;
this.minSequenceId = Long.MAX_VALUE;
this.memStoreLAB = memStoreLAB;
this.size = new AtomicLong(size);
this.size = new AtomicLong(0);
this.tagsPresent = false;
this.constantCellMetaDataSize = constantCellSize;
this.timeRangeTracker = new TimeRangeTracker();
}
@ -77,9 +80,8 @@ public abstract class Segment {
this.comparator = segment.getComparator();
this.minSequenceId = segment.getMinSequenceId();
this.memStoreLAB = segment.getMemStoreLAB();
this.size = new AtomicLong(segment.getSize());
this.size = new AtomicLong(segment.keySize());
this.tagsPresent = segment.isTagsPresent();
this.constantCellMetaDataSize = segment.getConstantCellMetaDataSize();
this.timeRangeTracker = segment.getTimeRangeTracker();
}
@ -100,7 +102,6 @@ public abstract class Segment {
}
/**
* Returns whether the segment has any cells
* @return whether the segment has any cells
*/
public boolean isEmpty() {
@ -108,7 +109,6 @@ public abstract class Segment {
}
/**
* Returns number of cells in segment
* @return number of cells in segment
*/
public int getCellsCount() {
@ -116,7 +116,6 @@ public abstract class Segment {
}
/**
* Returns the first cell in the segment that has equal or greater key than the given cell
* @return the first cell in the segment that has equal or greater key than the given cell
*/
public Cell getFirstAfter(Cell cell) {
@ -131,9 +130,8 @@ public abstract class Segment {
* Closing a segment before it is being discarded
*/
public void close() {
MemStoreLAB mslab = getMemStoreLAB();
if(mslab != null) {
mslab.close();
if (this.memStoreLAB != null) {
this.memStoreLAB.close();
}
// do not set MSLab to null as scanners may still be reading the data here and need to decrease
// the counter when they finish
@ -145,12 +143,12 @@ public abstract class Segment {
* @return either the given cell or its clone
*/
public Cell maybeCloneWithAllocator(Cell cell) {
if (getMemStoreLAB() == null) {
if (this.memStoreLAB == null) {
return cell;
}
int len = getCellLength(cell);
ByteRange alloc = getMemStoreLAB().allocateBytes(len);
ByteRange alloc = this.memStoreLAB.allocateBytes(len);
if (alloc == null) {
// The allocation was too large, allocator decided
// not to do anything with it.
@ -180,27 +178,17 @@ public abstract class Segment {
}
public void incScannerCount() {
if(getMemStoreLAB() != null) {
getMemStoreLAB().incScannerCount();
if (this.memStoreLAB != null) {
this.memStoreLAB.incScannerCount();
}
}
public void decScannerCount() {
if(getMemStoreLAB() != null) {
getMemStoreLAB().decScannerCount();
if (this.memStoreLAB != null) {
this.memStoreLAB.decScannerCount();
}
}
/**
* Setting the heap size of the segment - used to account for different class overheads
* @return this object
*/
public Segment setSize(long size) {
this.size.set(size);
return this;
}
/**
* Setting the CellSet of the segment - used only for flat immutable segment for setting
* immutable CellSet after its creation in immutable segment constructor
@ -212,22 +200,23 @@ public abstract class Segment {
return this;
}
/* return only cell's heap size */
public abstract long keySize();
/**
* @return Sum of all cell's size.
*/
public long keySize() {
return this.size.get();
}
/**
* Returns the heap size of the segment
* @return the heap size of the segment
*/
public long getSize() {
return size.get();
}
public abstract long size();
/**
* Updates the heap size counter of the segment by the given delta
*/
public void incSize(long delta) {
size.addAndGet(delta);
this.size.addAndGet(delta);
}
public long getMinSequenceId() {
@ -260,7 +249,6 @@ public abstract class Segment {
}
/**
* Returns a set of all cells in the segment
* @return a set of all cells in the segment
*/
protected CellSet getCellSet() {
@ -302,6 +290,11 @@ public abstract class Segment {
return s;
}
protected long heapSizeChange(Cell cell, boolean succ) {
return succ ? ClassSize
.align(ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY + CellUtil.estimatedHeapSizeOf(cell)) : 0;
}
/**
* Returns a subset of the segment cell set, which starts with the given cell
* @param firstCell a cell in the segment
@ -312,7 +305,7 @@ public abstract class Segment {
}
@VisibleForTesting
public MemStoreLAB getMemStoreLAB() {
MemStoreLAB getMemStoreLAB() {
return memStoreLAB;
}
@ -326,29 +319,13 @@ public abstract class Segment {
}
}
/*
* Calculate how the MemStore size has changed. Includes overhead of the
* backing Map.
* @param cell
* @param notPresent True if the cell was NOT present in the set.
* @return change in size
*/
protected long heapSizeChange(final Cell cell, final boolean notPresent){
return
notPresent ?
ClassSize.align(constantCellMetaDataSize + CellUtil.estimatedHeapSizeOf(cell)) : 0;
}
public long getConstantCellMetaDataSize() {
return this.constantCellMetaDataSize;
}
@Override
public String toString() {
String res = "Store segment of type "+this.getClass().getName()+"; ";
res += "isEmpty "+(isEmpty()?"yes":"no")+"; ";
res += "cellCount "+getCellsCount()+"; ";
res += "size "+getSize()+"; ";
res += "cellsSize "+keySize()+"; ";
res += "heapSize "+size()+"; ";
res += "Min ts "+getMinTimestamp()+"; ";
return res;
}

43
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SegmentFactory.java
View File

@ -39,6 +39,7 @@ public final class SegmentFactory {
private SegmentFactory() {}
private static SegmentFactory instance = new SegmentFactory();
public static SegmentFactory instance() {
return instance;
}
@ -46,47 +47,43 @@ 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) {
MemStoreLAB memStoreLAB = getMemStoreLAB(conf);
return
new ImmutableSegment(comparator, iterator, memStoreLAB);
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,
"wrong immutable segment type");
return new ImmutableSegment(comparator, iterator, getMemStoreLAB(conf), numOfCells,
segmentType);
}
// create empty immutable segment
public ImmutableSegment createImmutableSegment(CellComparator comparator, long size) {
MutableSegment segment = generateMutableSegment(null, comparator, null, size);
public ImmutableSegment createImmutableSegment(CellComparator comparator) {
MutableSegment segment = generateMutableSegment(null, comparator, null);
return createImmutableSegment(segment);
}
// create immutable segment from mutable
// create immutable segment from mutable segment
public ImmutableSegment createImmutableSegment(MutableSegment segment) {
return new ImmutableSegment(segment);
}
// create mutable segment
public MutableSegment createMutableSegment(final Configuration conf,
CellComparator comparator, long size) {
public MutableSegment createMutableSegment(final Configuration conf, CellComparator comparator) {
MemStoreLAB memStoreLAB = getMemStoreLAB(conf);
return generateMutableSegment(conf, comparator, memStoreLAB, size);
}
// 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, "wrong immutable segment type");
MemStoreLAB memStoreLAB = getMemStoreLAB(conf);
return
new ImmutableSegment(comparator, iterator, memStoreLAB, numOfCells, segmentType);
return generateMutableSegment(conf, comparator, memStoreLAB);
}
//****** private methods to instantiate concrete store segments **********//
private MutableSegment generateMutableSegment(
final Configuration conf, CellComparator comparator, MemStoreLAB memStoreLAB, long size) {
private MutableSegment generateMutableSegment(final Configuration conf, CellComparator comparator,
MemStoreLAB memStoreLAB) {
// TBD use configuration to set type of segment
CellSet set = new CellSet(comparator);
return new MutableSegment(set, comparator, memStoreLAB, size);
return new MutableSegment(set, comparator, memStoreLAB);
}
private MemStoreLAB getMemStoreLAB(Configuration conf) {

96
hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestHeapSize.java
View File

@ -40,6 +40,7 @@ import java.lang.management.ManagementFactory;
import java.lang.management.RuntimeMXBean;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
@ -49,6 +50,7 @@ import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static org.junit.Assert.assertEquals;
@ -310,19 +312,107 @@ public class TestHeapSize {
// DefaultMemStore Deep Overhead
actual = DefaultMemStore.DEEP_OVERHEAD;
expected = ClassSize.estimateBase(cl, false);
if (expected != actual) {
ClassSize.estimateBase(cl, true);
assertEquals(expected, actual);
}
// CompactingMemStore Deep Overhead
cl = CompactingMemStore.class;
actual = CompactingMemStore.DEEP_OVERHEAD;
expected = ClassSize.estimateBase(cl, false);
expected += ClassSize.estimateBase(AtomicBoolean.class, false);
expected += ClassSize.estimateBase(AtomicBoolean.class, false);
expected += ClassSize.estimateBase(CompactionPipeline.class, false);
expected += ClassSize.estimateBase(LinkedList.class, false);
expected += ClassSize.estimateBase(MemStoreCompactor.class, false);
expected += ClassSize.estimateBase(AtomicBoolean.class, false);
if (expected != actual) {
ClassSize.estimateBase(cl, true);
ClassSize.estimateBase(AtomicBoolean.class, true);
ClassSize.estimateBase(AtomicBoolean.class, true);
ClassSize.estimateBase(CompactionPipeline.class, true);
ClassSize.estimateBase(LinkedList.class, true);
ClassSize.estimateBase(MemStoreCompactor.class, true);
ClassSize.estimateBase(AtomicBoolean.class, true);
assertEquals(expected, actual);
}
// Segment Deep overhead
cl = Segment.class;
actual = Segment.DEEP_OVERHEAD;
expected = ClassSize.estimateBase(cl, false);
expected += ClassSize.estimateBase(AtomicLong.class, false);
expected += ClassSize.estimateBase(AtomicReference.class, false);
expected += ClassSize.estimateBase(CellSet.class, false);
expected += ClassSize.estimateBase(ConcurrentSkipListMap.class, false);
expected += ClassSize.estimateBase(TimeRangeTracker.class, false);
if(expected != actual) {
if (expected != actual) {
ClassSize.estimateBase(cl, true);
ClassSize.estimateBase(AtomicLong.class, true);
ClassSize.estimateBase(AtomicReference.class, true);
ClassSize.estimateBase(CellSet.class, true);
ClassSize.estimateBase(ConcurrentSkipListMap.class, true);
ClassSize.estimateBase(TimeRangeTracker.class, true);
assertEquals(expected, actual);
}
// MutableSegment Deep overhead
cl = MutableSegment.class;
actual = MutableSegment.DEEP_OVERHEAD;
expected = ClassSize.estimateBase(cl, false);
expected += ClassSize.estimateBase(AtomicLong.class, false);
expected += ClassSize.estimateBase(AtomicReference.class, false);
expected += ClassSize.estimateBase(CellSet.class, false);
expected += ClassSize.estimateBase(TimeRangeTracker.class, false);
expected += ClassSize.estimateBase(ConcurrentSkipListMap.class, false);
if (expected != actual) {
ClassSize.estimateBase(cl, true);
ClassSize.estimateBase(AtomicLong.class, true);
ClassSize.estimateBase(AtomicReference.class, true);
ClassSize.estimateBase(CellSet.class, true);
ClassSize.estimateBase(TimeRangeTracker.class, true);
ClassSize.estimateBase(ConcurrentSkipListMap.class, true);
assertEquals(expected, actual);
}
// ImmutableSegment Deep overhead
cl = ImmutableSegment.class;
actual = ImmutableSegment.DEEP_OVERHEAD_CSLM;
expected = ClassSize.estimateBase(cl, false);
expected += ClassSize.estimateBase(AtomicLong.class, false);
expected += ClassSize.estimateBase(AtomicReference.class, false);
expected += ClassSize.estimateBase(CellSet.class, false);
expected += ClassSize.estimateBase(TimeRangeTracker.class, false);
expected += ClassSize.estimateBase(TimeRange.class, false);
expected += ClassSize.estimateBase(ConcurrentSkipListMap.class, false);
if (expected != actual) {
ClassSize.estimateBase(cl, true);
ClassSize.estimateBase(AtomicLong.class, true);
ClassSize.estimateBase(AtomicReference.class, true);
ClassSize.estimateBase(CellSet.class, true);
ClassSize.estimateBase(TimeRangeTracker.class, true);
ClassSize.estimateBase(TimeRange.class, true);
ClassSize.estimateBase(ConcurrentSkipListMap.class, true);
assertEquals(expected, actual);
}
actual = ImmutableSegment.DEEP_OVERHEAD_CAM;
expected = ClassSize.estimateBase(cl, false);
expected += ClassSize.estimateBase(AtomicLong.class, false);
expected += ClassSize.estimateBase(AtomicReference.class, false);
expected += ClassSize.estimateBase(CellSet.class, false);
expected += ClassSize.estimateBase(TimeRangeTracker.class, false);
expected += ClassSize.estimateBase(TimeRange.class, false);
expected += ClassSize.estimateBase(CellArrayMap.class, false);
if (expected != actual) {
ClassSize.estimateBase(cl, true);
ClassSize.estimateBase(AtomicLong.class, true);
ClassSize.estimateBase(AtomicReference.class, true);
ClassSize.estimateBase(CellSet.class, true);
ClassSize.estimateBase(TimeRangeTracker.class, true);
ClassSize.estimateBase(TimeRange.class, true);
ClassSize.estimateBase(CellArrayMap.class, true);
assertEquals(expected, actual);
}
// Store Overhead
cl = HStore.class;
actual = HStore.FIXED_OVERHEAD;

22
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java
View File

@ -377,7 +377,7 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
private long runSnapshot(final AbstractMemStore hmc, boolean useForce)
throws IOException {
// Save off old state.
long oldHistorySize = hmc.getSnapshot().getSize();
long oldHistorySize = hmc.getSnapshot().keySize();
long prevTimeStamp = hmc.timeOfOldestEdit();
hmc.snapshot();
@ -547,10 +547,6 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
memstore.add(new KeyValue(row, fam, qf1, 3, val));
assertEquals(3, memstore.getActive().getCellsCount());
while (((CompactingMemStore)memstore).isMemStoreFlushingInMemory()) {
Threads.sleep(10);
}
assertTrue(chunkPool.getPoolSize() == 0);
// Chunks will be put back to pool after close scanners;
@ -597,9 +593,6 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
long size = memstore.getFlushableSize();
((CompactingMemStore)memstore).flushInMemory(); // push keys to pipeline and compact
while (((CompactingMemStore)memstore).isMemStoreFlushingInMemory()) {
Threads.sleep(10);
}
assertEquals(0, memstore.getSnapshot().getCellsCount());
assertEquals(264, regionServicesForStores.getGlobalMemstoreTotalSize());
@ -625,9 +618,6 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
long size = memstore.getFlushableSize();
((CompactingMemStore)memstore).flushInMemory(); // push keys to pipeline and compact
while (((CompactingMemStore)memstore).isMemStoreFlushingInMemory()) {
Threads.sleep(1000);
}
int counter = 0;
for ( Segment s : memstore.getSegments()) {
counter += s.getCellsCount();
@ -641,9 +631,6 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
size = memstore.getFlushableSize();
((CompactingMemStore)memstore).flushInMemory(); // push keys to pipeline and compact
while (((CompactingMemStore)memstore).isMemStoreFlushingInMemory()) {
Threads.sleep(10);
}
assertEquals(0, memstore.getSnapshot().getCellsCount());
assertEquals(368, regionServicesForStores.getGlobalMemstoreTotalSize());
@ -672,9 +659,6 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
String tstStr = "\n\nFlushable size after first flush in memory:" + size
+ ". Is MemmStore in compaction?:" + ((CompactingMemStore)memstore).isMemStoreFlushingInMemory();
while (((CompactingMemStore)memstore).isMemStoreFlushingInMemory()) {
Threads.sleep(10);
}
assertEquals(0, memstore.getSnapshot().getCellsCount());
assertEquals(264, regionServicesForStores.getGlobalMemstoreTotalSize());
@ -719,7 +703,7 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
private void addRowsByKeys(final AbstractMemStore hmc, String[] keys) {
byte[] fam = Bytes.toBytes("testfamily");
byte[] qf = Bytes.toBytes("testqualifier");
long size = hmc.getActive().getSize();
long size = hmc.getActive().keySize();
for (int i = 0; i < keys.length; i++) {
long timestamp = System.currentTimeMillis();
Threads.sleep(1); // to make sure each kv gets a different ts
@ -729,7 +713,7 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
hmc.add(kv);
LOG.debug("added kv: " + kv.getKeyString() + ", timestamp:" + kv.getTimestamp());
}
regionServicesForStores.addAndGetGlobalMemstoreSize(hmc.getActive().getSize() - size);
regionServicesForStores.addAndGetGlobalMemstoreSize(hmc.getActive().keySize() - size);
}
private class EnvironmentEdgeForMemstoreTest implements EnvironmentEdge {

4
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellArrayMapMemStore.java
View File

@ -333,7 +333,7 @@ public class TestCompactingToCellArrayMapMemStore extends TestCompactingMemStore
private void addRowsByKeys(final AbstractMemStore hmc, String[] keys) {
byte[] fam = Bytes.toBytes("testfamily");
byte[] qf = Bytes.toBytes("testqualifier");
long size = hmc.getActive().getSize();//
long size = hmc.getActive().size();//
for (int i = 0; i < keys.length; i++) {
long timestamp = System.currentTimeMillis();
Threads.sleep(1); // to make sure each kv gets a different ts
@ -343,7 +343,7 @@ public class TestCompactingToCellArrayMapMemStore extends TestCompactingMemStore
hmc.add(kv);
LOG.debug("added kv: " + kv.getKeyString() + ", timestamp" + kv.getTimestamp());
}
regionServicesForStores.addAndGetGlobalMemstoreSize(hmc.getActive().getSize() - size);//
regionServicesForStores.addAndGetGlobalMemstoreSize(hmc.getActive().size() - size);//
}
private class EnvironmentEdgeForMemstoreTest implements EnvironmentEdge {

49
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPerColumnFamilyFlush.java
View File

@ -172,8 +172,9 @@ public class TestPerColumnFamilyFlush {
// The total memstore size should be the same as the sum of the sizes of
// memstores of CF1, CF2 and CF3.
assertEquals(totalMemstoreSize + 3 * DefaultMemStore.DEEP_OVERHEAD, cf1MemstoreSize
+ cf2MemstoreSize + cf3MemstoreSize);
assertEquals(
totalMemstoreSize + (3 * (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD)),
cf1MemstoreSize + cf2MemstoreSize + cf3MemstoreSize);
// Flush!
region.flush(false);
@ -192,7 +193,7 @@ public class TestPerColumnFamilyFlush {
// We should have cleared out only CF1, since we chose the flush thresholds
// and number of puts accordingly.
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf1MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD, cf1MemstoreSize);
// Nothing should have happened to CF2, ...
assertEquals(cf2MemstoreSize, oldCF2MemstoreSize);
// ... or CF3
@ -201,8 +202,9 @@ public class TestPerColumnFamilyFlush {
// LSN in the memstore of CF2.
assertEquals(smallestSeqInRegionCurrentMemstore, smallestSeqCF2);
// Of course, this should hold too.
assertEquals(totalMemstoreSize + 2 * DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSize
+ cf3MemstoreSize);
assertEquals(
totalMemstoreSize + (2 * (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD)),
cf2MemstoreSize + cf3MemstoreSize);
// Now add more puts (mostly for CF2), so that we only flush CF2 this time.
for (int i = 1200; i < 2400; i++) {
@ -229,11 +231,12 @@ public class TestPerColumnFamilyFlush {
.getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
// CF1 and CF2, both should be absent.
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf1MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD, cf1MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD, cf2MemstoreSize);
// CF3 shouldn't have been touched.
assertEquals(cf3MemstoreSize, oldCF3MemstoreSize);
assertEquals(totalMemstoreSize + DefaultMemStore.DEEP_OVERHEAD, cf3MemstoreSize);
assertEquals(totalMemstoreSize + (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD),
cf3MemstoreSize);
assertEquals(smallestSeqInRegionCurrentMemstore, smallestSeqCF3);
// What happens when we hit the memstore limit, but we are not able to find
@ -296,8 +299,9 @@ public class TestPerColumnFamilyFlush {
// The total memstore size should be the same as the sum of the sizes of
// memstores of CF1, CF2 and CF3.
assertEquals(totalMemstoreSize + 3 * DefaultMemStore.DEEP_OVERHEAD, cf1MemstoreSize
+ cf2MemstoreSize + cf3MemstoreSize);
assertEquals(
totalMemstoreSize + (3 * (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD)),
cf1MemstoreSize + cf2MemstoreSize + cf3MemstoreSize);
// Flush!
region.flush(false);
@ -310,9 +314,9 @@ public class TestPerColumnFamilyFlush {
region.getWAL().getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
// Everything should have been cleared
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf1MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf3MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD, cf1MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD, cf2MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD, cf3MemstoreSize);
assertEquals(0, totalMemstoreSize);
assertEquals(HConstants.NO_SEQNUM, smallestSeqInRegionCurrentMemstore);
HBaseTestingUtility.closeRegionAndWAL(region);
@ -379,12 +383,13 @@ public class TestPerColumnFamilyFlush {
cf3MemstoreSize = desiredRegion.getStore(FAMILY3).getMemStoreSize();
// CF1 Should have been flushed
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf1MemstoreSize);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD, cf1MemstoreSize);
// CF2 and CF3 shouldn't have been flushed.
assertTrue(cf2MemstoreSize > 0);
assertTrue(cf3MemstoreSize > 0);
assertEquals(totalMemstoreSize + 2 * DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSize
+ cf3MemstoreSize);
assertEquals(
totalMemstoreSize + (2 * (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD)),
cf2MemstoreSize + cf3MemstoreSize);
// Wait for the RS report to go across to the master, so that the master
// is aware of which sequence ids have been flushed, before we kill the RS.
@ -521,12 +526,12 @@ public class TestPerColumnFamilyFlush {
});
LOG.info("Finished waiting on flush after too many WALs...");
// Individual families should have been flushed.
assertEquals(DefaultMemStore.DEEP_OVERHEAD,
desiredRegion.getStore(FAMILY1).getMemStoreSize());
assertEquals(DefaultMemStore.DEEP_OVERHEAD,
desiredRegion.getStore(FAMILY2).getMemStoreSize());
assertEquals(DefaultMemStore.DEEP_OVERHEAD,
desiredRegion.getStore(FAMILY3).getMemStoreSize());
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
desiredRegion.getStore(FAMILY1).getMemStoreSize());
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
desiredRegion.getStore(FAMILY2).getMemStoreSize());
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
desiredRegion.getStore(FAMILY3).getMemStoreSize());
// let WAL cleanOldLogs
assertNull(getWAL(desiredRegion).rollWriter(true));
assertTrue(getNumRolledLogFiles(desiredRegion) < maxLogs);

129
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
View File

@ -19,10 +19,7 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
@ -31,19 +28,14 @@ import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MiniHBaseCluster;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.regionserver.wal.FSHLog;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.testclassification.RegionServerTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.JVMClusterUtil;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.wal.WAL;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@ -59,7 +51,6 @@ import static org.junit.Assert.assertTrue;
@Category({ RegionServerTests.class, LargeTests.class })
public class TestWalAndCompactingMemStoreFlush {
private static final Log LOG = LogFactory.getLog(TestWalAndCompactingMemStoreFlush.class);
private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
private static final Path DIR = TEST_UTIL.getDataTestDir("TestHRegion");
public static final TableName TABLENAME = TableName.valueOf("TestWalAndCompactingMemStoreFlush",
@ -201,12 +192,13 @@ public class TestWalAndCompactingMemStoreFlush {
// memstores of CF1, CF2 and CF3.
String msg = "totalMemstoreSize="+totalMemstoreSize +
" DefaultMemStore.DEEP_OVERHEAD="+DefaultMemStore.DEEP_OVERHEAD +
" DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM="+CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM
+
" CompactingMemStore.DEEP_OVERHEAD="+CompactingMemStore.DEEP_OVERHEAD +
" cf1MemstoreSizePhaseI="+cf1MemstoreSizePhaseI +
" cf2MemstoreSizePhaseI="+cf2MemstoreSizePhaseI +
" cf3MemstoreSizePhaseI="+cf3MemstoreSizePhaseI ;
assertEquals(msg,totalMemstoreSize + 3 * DefaultMemStore.DEEP_OVERHEAD,
assertEquals(msg,
totalMemstoreSize + 2 * (CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD)
+ (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD),
cf1MemstoreSizePhaseI + cf2MemstoreSizePhaseI + cf3MemstoreSizePhaseI);
// Flush!!!!!!!!!!!!!!!!!!!!!!
@ -220,11 +212,6 @@ public class TestWalAndCompactingMemStoreFlush {
((CompactingMemStore) region.getStore(FAMILY3).getMemStore()).flushInMemory();
region.flush(false);
// CF3 should be compacted so wait here to be sure the compaction is done
while (((CompactingMemStore) region.getStore(FAMILY3).getMemStore())
.isMemStoreFlushingInMemory())
Threads.sleep(10);
// Recalculate everything
long cf1MemstoreSizePhaseII = region.getStore(FAMILY1).getMemStoreSize();
long cf2MemstoreSizePhaseII = region.getStore(FAMILY2).getMemStoreSize();
@ -239,8 +226,6 @@ public class TestWalAndCompactingMemStoreFlush {
s = s + "DefaultMemStore DEEP_OVERHEAD is:" + DefaultMemStore.DEEP_OVERHEAD
+ ", CompactingMemStore DEEP_OVERHEAD is:" + CompactingMemStore.DEEP_OVERHEAD
+ ", CompactingMemStore DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM is:" + CompactingMemStore
.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM
+ "\n----After first flush! CF1 should be flushed to memory, but not compacted.---\n"
+ "Size of CF1 is:" + cf1MemstoreSizePhaseII + ", size of CF2 is:" + cf2MemstoreSizePhaseII
+ ", size of CF3 is:" + cf3MemstoreSizePhaseII + "\n";
@ -249,12 +234,13 @@ public class TestWalAndCompactingMemStoreFlush {
assertTrue(cf1MemstoreSizePhaseII < cf1MemstoreSizePhaseI);
// CF2 should become empty
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSizePhaseII);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf2MemstoreSizePhaseII);
// verify that CF3 was flushed to memory and was compacted (this is approximation check)
assertTrue(cf3MemstoreSizePhaseI/2+DefaultMemStore.DEEP_OVERHEAD +
CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM >
cf3MemstoreSizePhaseII);
assertTrue(cf3MemstoreSizePhaseI / 2 + CompactingMemStore.DEEP_OVERHEAD
+ ImmutableSegment.DEEP_OVERHEAD_CAM
+ CompactionPipeline.ENTRY_OVERHEAD > cf3MemstoreSizePhaseII);
// CF3 was compacted and flattened!
assertTrue("\n<<< Size of CF3 in phase I - " + cf3MemstoreSizePhaseI
@ -315,7 +301,8 @@ public class TestWalAndCompactingMemStoreFlush {
// CF1's pipeline component (inserted before first flush) should be flushed to disk
// CF2 should be flushed to disk
assertTrue(cf1MemstoreSizePhaseIII > cf1MemstoreSizePhaseIV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSizePhaseIV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf2MemstoreSizePhaseIV);
// CF3 shouldn't have been touched.
assertEquals(cf3MemstoreSizePhaseIV, cf3MemstoreSizePhaseII);
@ -340,12 +327,16 @@ public class TestWalAndCompactingMemStoreFlush {
long smallestSeqInRegionCurrentMemstorePhaseV = getWAL(region)
.getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
assertTrue(DefaultMemStore.DEEP_OVERHEAD < cf1MemstoreSizePhaseV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSizePhaseV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf3MemstoreSizePhaseV);
assertTrue(
CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD < cf1MemstoreSizePhaseV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf2MemstoreSizePhaseV);
assertEquals(CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf3MemstoreSizePhaseV);
region.flush(true); // flush once again in order to be sure that everything is empty
assertEquals(DefaultMemStore.DEEP_OVERHEAD, region.getStore(FAMILY1).getMemStoreSize());
assertEquals(CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
region.getStore(FAMILY1).getMemStoreSize());
// What happens when we hit the memstore limit, but we are not able to find
// any Column Family above the threshold?
@ -453,12 +444,12 @@ public class TestWalAndCompactingMemStoreFlush {
// memstores of CF1, CF2 and CF3.
String msg = "totalMemstoreSize="+totalMemstoreSize +
" DefaultMemStore.DEEP_OVERHEAD="+DefaultMemStore.DEEP_OVERHEAD +
" DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM="+CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM
+
" cf1MemstoreSizePhaseI="+cf1MemstoreSizePhaseI +
" cf2MemstoreSizePhaseI="+cf2MemstoreSizePhaseI +
" cf3MemstoreSizePhaseI="+cf3MemstoreSizePhaseI ;
assertEquals(msg,totalMemstoreSize + 3 * DefaultMemStore.DEEP_OVERHEAD,
assertEquals(msg,
totalMemstoreSize + 2 * (CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD)
+ (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD),
cf1MemstoreSizePhaseI + cf2MemstoreSizePhaseI + cf3MemstoreSizePhaseI);
// Flush!!!!!!!!!!!!!!!!!!!!!!
@ -472,11 +463,6 @@ public class TestWalAndCompactingMemStoreFlush {
((CompactingMemStore) region.getStore(FAMILY3).getMemStore()).flushInMemory();
region.flush(false);
// CF3 should be compacted so wait here to be sure the compaction is done
while (((CompactingMemStore) region.getStore(FAMILY3).getMemStore())
.isMemStoreFlushingInMemory())
Threads.sleep(10);
// Recalculate everything
long cf1MemstoreSizePhaseII = region.getStore(FAMILY1).getMemStoreSize();
long cf2MemstoreSizePhaseII = region.getStore(FAMILY2).getMemStoreSize();
@ -491,24 +477,23 @@ public class TestWalAndCompactingMemStoreFlush {
s = s + "DefaultMemStore DEEP_OVERHEAD is:" + DefaultMemStore.DEEP_OVERHEAD
+ ", CompactingMemStore DEEP_OVERHEAD is:" + CompactingMemStore.DEEP_OVERHEAD
+ ", CompactingMemStore DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM is:" + CompactingMemStore
.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM
+ "\n----After first flush! CF1 should be flushed to memory, but not compacted.---\n"
+ "Size of CF1 is:" + cf1MemstoreSizePhaseII + ", size of CF2 is:" + cf2MemstoreSizePhaseII
+ ", size of CF3 is:" + cf3MemstoreSizePhaseII + "\n";
// CF1 was flushed to memory, but there is nothing to compact, should
// remain the same size plus renewed empty skip-list
assertEquals(s, cf1MemstoreSizePhaseII,
cf1MemstoreSizePhaseI + CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM);
assertEquals(s, cf1MemstoreSizePhaseII, cf1MemstoreSizePhaseI
+ ImmutableSegment.DEEP_OVERHEAD_CAM + CompactionPipeline.ENTRY_OVERHEAD);
// CF2 should become empty
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSizePhaseII);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf2MemstoreSizePhaseII);
// verify that CF3 was flushed to memory and was compacted (this is approximation check)
assertTrue(cf3MemstoreSizePhaseI/2+DefaultMemStore.DEEP_OVERHEAD +
CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM >
cf3MemstoreSizePhaseII);
assertTrue(cf3MemstoreSizePhaseI / 2 + CompactingMemStore.DEEP_OVERHEAD
+ ImmutableSegment.DEEP_OVERHEAD_CAM
+ CompactionPipeline.ENTRY_OVERHEAD > cf3MemstoreSizePhaseII);
assertTrue(cf3MemstoreSizePhaseI/2 < cf3MemstoreSizePhaseII);
@ -564,9 +549,10 @@ public class TestWalAndCompactingMemStoreFlush {
// CF1's pipeline component (inserted before first flush) should be flushed to disk
// CF2 should be flushed to disk
assertEquals(cf1MemstoreSizePhaseIII - cf1MemstoreSizePhaseI + DefaultMemStore.DEEP_OVERHEAD,
cf1MemstoreSizePhaseIV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSizePhaseIV);
assertEquals(cf1MemstoreSizePhaseIII - cf1MemstoreSizePhaseI + CompactingMemStore.DEEP_OVERHEAD
+ MutableSegment.DEEP_OVERHEAD, cf1MemstoreSizePhaseIV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf2MemstoreSizePhaseIV);
// CF3 shouldn't have been touched.
assertEquals(cf3MemstoreSizePhaseIV, cf3MemstoreSizePhaseII);
@ -590,9 +576,12 @@ public class TestWalAndCompactingMemStoreFlush {
long smallestSeqInRegionCurrentMemstorePhaseV = getWAL(region)
.getEarliestMemstoreSeqNum(region.getRegionInfo().getEncodedNameAsBytes());
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf1MemstoreSizePhaseV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSizePhaseV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf3MemstoreSizePhaseV);
assertEquals(CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf1MemstoreSizePhaseV);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf2MemstoreSizePhaseV);
assertEquals(CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf3MemstoreSizePhaseV);
// Because there is nothing in any memstore the WAL's LSN should be -1
assertEquals(smallestSeqInRegionCurrentMemstorePhaseV, HConstants.NO_SEQNUM);
@ -672,24 +661,17 @@ public class TestWalAndCompactingMemStoreFlush {
// memstores of CF1, CF2 and CF3.
String msg = "totalMemstoreSize="+totalMemstoreSize +
" DefaultMemStore.DEEP_OVERHEAD="+DefaultMemStore.DEEP_OVERHEAD +
" DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM="+CompactingMemStore.DEEP_OVERHEAD_PER_PIPELINE_SKIPLIST_ITEM
+
" cf1MemstoreSizePhaseI="+cf1MemstoreSizePhaseI +
" cf2MemstoreSizePhaseI="+cf2MemstoreSizePhaseI +
" cf3MemstoreSizePhaseI="+cf3MemstoreSizePhaseI ;
assertEquals(msg, totalMemstoreSize + 3 * DefaultMemStore.DEEP_OVERHEAD,
assertEquals(msg,
totalMemstoreSize + 2 * (CompactingMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD)
+ (DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD),
cf1MemstoreSizePhaseI + cf2MemstoreSizePhaseI + cf3MemstoreSizePhaseI);
// Flush!
((CompactingMemStore) region.getStore(FAMILY1).getMemStore()).flushInMemory();
((CompactingMemStore) region.getStore(FAMILY3).getMemStore()).flushInMemory();
// CF1 and CF3 should be compacted so wait here to be sure the compaction is done
while (((CompactingMemStore) region.getStore(FAMILY1).getMemStore())
.isMemStoreFlushingInMemory())
Threads.sleep(10);
while (((CompactingMemStore) region.getStore(FAMILY3).getMemStore())
.isMemStoreFlushingInMemory())
Threads.sleep(10);
region.flush(false);
long cf2MemstoreSizePhaseII = region.getStore(FAMILY2).getMemStoreSize();
@ -701,7 +683,8 @@ public class TestWalAndCompactingMemStoreFlush {
long smallestSeqCF3PhaseII = region.getOldestSeqIdOfStore(FAMILY3);
// CF2 should have been cleared
assertEquals(DefaultMemStore.DEEP_OVERHEAD, cf2MemstoreSizePhaseII);
assertEquals(DefaultMemStore.DEEP_OVERHEAD + MutableSegment.DEEP_OVERHEAD,
cf2MemstoreSizePhaseII);
String s = "\n\n----------------------------------\n"
+ "Upon initial insert and flush, LSN of CF1 is:"
@ -739,13 +722,6 @@ public class TestWalAndCompactingMemStoreFlush {
// Flush!
((CompactingMemStore) region.getStore(FAMILY1).getMemStore()).flushInMemory();
((CompactingMemStore) region.getStore(FAMILY3).getMemStore()).flushInMemory();
// CF1 and CF3 should be compacted so wait here to be sure the compaction is done
while (((CompactingMemStore) region.getStore(FAMILY1).getMemStore())
.isMemStoreFlushingInMemory())
Threads.sleep(10);
while (((CompactingMemStore) region.getStore(FAMILY3).getMemStore())
.isMemStoreFlushingInMemory())
Threads.sleep(10);
region.flush(false);
long smallestSeqInRegionCurrentMemstorePhaseIV =
@ -768,24 +744,7 @@ public class TestWalAndCompactingMemStoreFlush {
HBaseTestingUtility.closeRegionAndWAL(region);
}
// Find the (first) region which has the specified name.
private static Pair<Region, HRegionServer> getRegionWithName(TableName tableName) {
MiniHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
List<JVMClusterUtil.RegionServerThread> rsts = cluster.getRegionServerThreads();
for (int i = 0; i < cluster.getRegionServerThreads().size(); i++) {
HRegionServer hrs = rsts.get(i).getRegionServer();
for (Region region : hrs.getOnlineRegions(tableName)) {
return Pair.newPair(region, hrs);
}
}
return null;
}
private WAL getWAL(Region region) {
return ((HRegion)region).getWAL();
}
private int getNumRolledLogFiles(Region region) {
return ((FSHLog)getWAL(region)).getNumRolledLogFiles();
}
}