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/09/15 12:38:04 UTC
[1/7] hbase git commit: Adding checks in Scanner's setStartRow and
setStopRow for invalid row key sizes.
Repository: hbase
Updated Branches:
refs/heads/master 1d6c90b49 -> 56be3ac7c
Adding checks in Scanner's setStartRow and setStopRow for invalid row key sizes.
Signed-off-by: Gary Helmling <ga...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/1cdc5acf
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/1cdc5acf
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/1cdc5acf
Branch: refs/heads/master
Commit: 1cdc5acfd456688f106287c19ae9af62fd03a3af
Parents: 8855670
Author: Dustin Pho <ph...@fb.com>
Authored: Mon Sep 12 13:25:02 2016 -0700
Committer: Gary Helmling <ga...@apache.org>
Committed: Mon Sep 12 16:54:51 2016 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hbase/client/Scan.java | 16 +++++++++++
.../apache/hadoop/hbase/client/TestScan.java | 28 ++++++++++++++++++++
2 files changed, 44 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/1cdc5acf/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java
index ee3ed43..22f611a 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java
@@ -373,8 +373,16 @@ public class Scan extends Query {
* next closest row after the specified row.
* @param startRow row to start scanner at or after
* @return this
+ * @throws IllegalArgumentException if startRow does not meet criteria
+ * for a row key (when length exceeds {@link HConstants#MAX_ROW_LENGTH})
*/
public Scan setStartRow(byte [] startRow) {
+ if (Bytes.len(startRow) > HConstants.MAX_ROW_LENGTH) {
+ throw new IllegalArgumentException(
+ "startRow's length must be less than or equal to " +
+ HConstants.MAX_ROW_LENGTH + " to meet the criteria" +
+ " for a row key.");
+ }
this.startRow = startRow;
return this;
}
@@ -389,8 +397,16 @@ public class Scan extends Query {
* use {@link #setRowPrefixFilter(byte[])}.
* The 'trailing 0' will not yield the desired result.</p>
* @return this
+ * @throws IllegalArgumentException if stopRow does not meet criteria
+ * for a row key (when length exceeds {@link HConstants#MAX_ROW_LENGTH})
*/
public Scan setStopRow(byte [] stopRow) {
+ if (Bytes.len(stopRow) > HConstants.MAX_ROW_LENGTH) {
+ throw new IllegalArgumentException(
+ "stopRow's length must be less than or equal to " +
+ HConstants.MAX_ROW_LENGTH + " to meet the criteria" +
+ " for a row key.");
+ }
this.stopRow = stopRow;
return this;
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/1cdc5acf/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestScan.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestScan.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestScan.java
index 129914f..16c74df 100644
--- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestScan.java
+++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestScan.java
@@ -25,6 +25,8 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Set;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.exceptions.IllegalArgumentIOException;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
import org.apache.hadoop.hbase.security.visibility.Authorizations;
@@ -132,5 +134,31 @@ public class TestScan {
fail("should not throw exception");
}
}
+
+ @Test
+ public void testSetStartRowAndSetStopRow() {
+ Scan scan = new Scan();
+ scan.setStartRow(null);
+ scan.setStartRow(new byte[1]);
+ scan.setStartRow(new byte[HConstants.MAX_ROW_LENGTH]);
+ try {
+ scan.setStartRow(new byte[HConstants.MAX_ROW_LENGTH+1]);
+ fail("should've thrown exception");
+ } catch (IllegalArgumentException iae) {
+ } catch (Exception e) {
+ fail("expected IllegalArgumentException to be thrown");
+ }
+
+ scan.setStopRow(null);
+ scan.setStopRow(new byte[1]);
+ scan.setStopRow(new byte[HConstants.MAX_ROW_LENGTH]);
+ try {
+ scan.setStopRow(new byte[HConstants.MAX_ROW_LENGTH+1]);
+ fail("should've thrown exception");
+ } catch (IllegalArgumentException iae) {
+ } catch (Exception e) {
+ fail("expected IllegalArgumentException to be thrown");
+ }
+ }
}
[7/7] hbase git commit: HBASE-16626 User customized RegionScanner
from 1.X is incompatible with 2.0.0's off-heap part. (Charlie Qiangeng Xu)
Posted by an...@apache.org.
HBASE-16626 User customized RegionScanner from 1.X is incompatible with 2.0.0's off-heap part. (Charlie Qiangeng Xu)
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/56be3ac7
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/56be3ac7
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/56be3ac7
Branch: refs/heads/master
Commit: 56be3ac7c8d203710f844cd799f2fa3496f0515a
Parents: e6f8f6d
Author: anoopsamjohn <an...@gmail.com>
Authored: Thu Sep 15 18:07:43 2016 +0530
Committer: anoopsamjohn <an...@gmail.com>
Committed: Thu Sep 15 18:07:43 2016 +0530
----------------------------------------------------------------------
.../org/apache/hadoop/hbase/regionserver/RegionScanner.java | 9 +++++++++
1 file changed, 9 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/56be3ac7/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java
index 5b33db4..4f9732d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionScanner.java
@@ -115,4 +115,13 @@ public interface RegionScanner extends InternalScanner, Shipper {
*/
boolean nextRaw(List<Cell> result, ScannerContext scannerContext)
throws IOException;
+
+ /**
+ * Empty implementation to provide compatibility for user migrating from 1.X
+ * @see <a href="https://issues.apache.org/jira/browse/HBASE-16626">HBASE-16626</a>
+ */
+ @Override
+ default void shipped() throws IOException {
+ // do nothing
+ }
}
[5/7] hbase git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/hbase
Posted by an...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/hbase
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/156a8b2b
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/156a8b2b
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/156a8b2b
Branch: refs/heads/master
Commit: 156a8b2bad2577ace5d0958acc79f56b2bc23e0c
Parents: 2ab3384 77b3273
Author: anoopsamjohn <an...@gmail.com>
Authored: Tue Sep 13 12:08:36 2016 +0530
Committer: anoopsamjohn <an...@gmail.com>
Committed: Tue Sep 13 12:08:36 2016 +0530
----------------------------------------------------------------------
----------------------------------------------------------------------
[4/7] hbase git commit: HBASE-16229 Cleaning up size and heapSize
calculation.
Posted by an...@apache.org.
HBASE-16229 Cleaning up size and heapSize calculation.
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/2ab33846
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/2ab33846
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/2ab33846
Branch: refs/heads/master
Commit: 2ab3384669f1df514de250918aeaa51e9ba6b71c
Parents: 831fb3c
Author: anoopsamjohn <an...@gmail.com>
Authored: Tue Sep 13 11:43:26 2016 +0530
Committer: anoopsamjohn <an...@gmail.com>
Committed: Tue Sep 13 11:43:26 2016 +0530
----------------------------------------------------------------------
.../org/apache/hadoop/hbase/util/ClassSize.java | 15 +++
.../hbase/regionserver/AbstractMemStore.java | 49 +++----
.../hbase/regionserver/CompactingMemStore.java | 92 ++++++-------
.../hbase/regionserver/CompactionPipeline.java | 31 +++--
.../hbase/regionserver/DefaultMemStore.java | 30 ++---
.../hbase/regionserver/ImmutableSegment.java | 84 ++++++------
.../hadoop/hbase/regionserver/MemStore.java | 6 +-
.../hbase/regionserver/MemStoreCompactor.java | 24 ++--
.../hbase/regionserver/MemStoreSnapshot.java | 2 +-
.../hbase/regionserver/MutableSegment.java | 25 ++--
.../hadoop/hbase/regionserver/Segment.java | 103 ++++++---------
.../hbase/regionserver/SegmentFactory.java | 43 +++----
.../apache/hadoop/hbase/io/TestHeapSize.java | 92 ++++++++++++-
.../regionserver/TestCompactingMemStore.java | 22 +---
.../TestCompactingToCellArrayMapMemStore.java | 4 +-
.../regionserver/TestPerColumnFamilyFlush.java | 49 +++----
.../TestWalAndCompactingMemStoreFlush.java | 129 +++++++------------
17 files changed, 417 insertions(+), 383 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ClassSize.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ClassSize.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ClassSize.java
index ff9dbcb..85a6483 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ClassSize.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ClassSize.java
@@ -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);
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java
index 419b76a..5e9f632 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java
@@ -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;
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/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 504ddab..177f222 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
@@ -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");
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/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 e0ba8c3..6a13f43 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
@@ -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);
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
index 50ab06b..b448b04 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java
@@ -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;
}
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/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 28f14d5..12b7916 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
@@ -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);
}
-
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
index 52ae6a3..d52b863 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStore.java
@@ -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();
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/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 470dc9c..714ffe3 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
@@ -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;
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSnapshot.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSnapshot.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSnapshot.java
index f64979f..1bb4511 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSnapshot.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSnapshot.java
@@ -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();
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MutableSegment.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MutableSegment.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MutableSegment.java
index 1cac7fd..3fb9723 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MutableSegment.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MutableSegment.java
@@ -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;
}
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java
index b4d1d61..01f3da9 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java
@@ -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,28 +178,18 @@ 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
* @return this object
@@ -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;
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/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 6351f13..510ebbd 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
@@ -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) {
- 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");
+ public MutableSegment createMutableSegment(final Configuration conf, CellComparator comparator) {
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) {
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestHeapSize.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestHeapSize.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestHeapSize.java
index e7d6661..6e8f831 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestHeapSize.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestHeapSize.java
@@ -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,16 +312,104 @@ 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(TimeRangeTracker.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);
+ 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);
- if(expected != actual) {
+ 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);
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/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 db0205e..211a6d8 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
@@ -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 {
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/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 1933343..fefe2c1 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
@@ -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 {
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/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 1615b99..6bfaa59 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
@@ -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);
[2/7] hbase git commit: HBASE-16592 Unify Delete request with AP
Posted by an...@apache.org.
HBASE-16592 Unify Delete request with AP
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/831fb3cc
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/831fb3cc
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/831fb3cc
Branch: refs/heads/master
Commit: 831fb3ccb8a0ba449d249962379afd268e8fe032
Parents: 1cdc5ac
Author: chenheng <ch...@apache.org>
Authored: Tue Sep 13 10:07:45 2016 +0800
Committer: chenheng <ch...@apache.org>
Committed: Tue Sep 13 10:07:45 2016 +0800
----------------------------------------------------------------------
.../hadoop/hbase/client/AbstractResponse.java | 38 ++++++++++++
.../hadoop/hbase/client/AsyncProcess.java | 22 ++++---
.../org/apache/hadoop/hbase/client/HTable.java | 43 ++++++++-----
.../hadoop/hbase/client/MultiResponse.java | 7 ++-
.../hadoop/hbase/client/SingleResponse.java | 65 ++++++++++++++++++++
.../hbase/protobuf/ResponseConverter.java | 14 +++++
.../hadoop/hbase/client/TestAsyncProcess.java | 18 +++---
.../hadoop/hbase/client/TestFromClientSide.java | 46 ++++++++++++++
8 files changed, 222 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/831fb3cc/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractResponse.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractResponse.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractResponse.java
new file mode 100644
index 0000000..7878d05
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractResponse.java
@@ -0,0 +1,38 @@
+/*
+ *
+ * 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.client;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+
+/**
+ * This class is used to extend AP to process single action request, like delete, get etc.
+ */
+@InterfaceAudience.Private
+abstract class AbstractResponse {
+
+ public enum ResponseType {
+
+ SINGLE (0),
+ MULTI (1);
+
+ ResponseType(int value) {}
+ }
+
+ public abstract ResponseType type();
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/831fb3cc/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
index c5745e9..1531201 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
@@ -756,14 +756,14 @@ class AsyncProcess {
@Override
public void run() {
- MultiResponse res;
+ AbstractResponse res;
CancellableRegionServerCallable callable = currentCallable;
try {
// setup the callable based on the actions, if we don't have one already from the request
if (callable == null) {
callable = createCallable(server, tableName, multiAction);
}
- RpcRetryingCaller<MultiResponse> caller = createCaller(callable);
+ RpcRetryingCaller<AbstractResponse> caller = createCaller(callable);
try {
if (callsInProgress != null) {
callsInProgress.add(callable);
@@ -785,9 +785,16 @@ class AsyncProcess {
receiveGlobalFailure(multiAction, server, numAttempt, t);
return;
}
-
- // Normal case: we received an answer from the server, and it's not an exception.
- receiveMultiAction(multiAction, server, res, numAttempt);
+ if (res.type() == AbstractResponse.ResponseType.MULTI) {
+ // Normal case: we received an answer from the server, and it's not an exception.
+ receiveMultiAction(multiAction, server, (MultiResponse) res, numAttempt);
+ } else {
+ if (results != null) {
+ SingleResponse singleResponse = (SingleResponse) res;
+ results[0] = singleResponse.getEntry();
+ }
+ decActionCounter(1);
+ }
} catch (Throwable t) {
// Something really bad happened. We are on the send thread that will now die.
LOG.error("Internal AsyncProcess #" + id + " error for "
@@ -1782,8 +1789,9 @@ class AsyncProcess {
* Create a caller. Isolated to be easily overridden in the tests.
*/
@VisibleForTesting
- protected RpcRetryingCaller<MultiResponse> createCaller(CancellableRegionServerCallable callable) {
- return rpcCallerFactory.<MultiResponse> newCaller();
+ protected RpcRetryingCaller<AbstractResponse> createCaller(
+ CancellableRegionServerCallable callable) {
+ return rpcCallerFactory.<AbstractResponse> newCaller();
}
@VisibleForTesting
http://git-wip-us.apache.org/repos/asf/hbase/blob/831fb3cc/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java
index 0d1b156..bcbb1da 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java
@@ -50,7 +50,6 @@ import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
-import org.apache.hadoop.hbase.ipc.HBaseRpcController;
import org.apache.hadoop.hbase.ipc.RegionCoprocessorRpcChannel;
import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
@@ -524,18 +523,25 @@ public class HTable implements Table {
@Override
public void delete(final Delete delete)
throws IOException {
- RegionServerCallable<Boolean> callable = new RegionServerCallable<Boolean>(connection,
- this.rpcControllerFactory, getName(), delete.getRow()) {
+ CancellableRegionServerCallable<SingleResponse> callable =
+ new CancellableRegionServerCallable<SingleResponse>(
+ connection, getName(), delete.getRow(), this.rpcControllerFactory) {
@Override
- protected Boolean rpcCall() throws Exception {
+ protected SingleResponse rpcCall() throws Exception {
MutateRequest request = RequestConverter.buildMutateRequest(
getLocation().getRegionInfo().getRegionName(), delete);
MutateResponse response = getStub().mutate(getRpcController(), request);
- return Boolean.valueOf(response.getProcessed());
+ return ResponseConverter.getResult(request, response, getRpcControllerCellScanner());
}
};
- rpcCallerFactory.<Boolean> newCaller(writeRpcTimeout).callWithRetries(callable,
- this.operationTimeout);
+ List<Row> rows = new ArrayList<Row>();
+ rows.add(delete);
+ AsyncRequestFuture ars = multiAp.submitAll(pool, tableName, rows,
+ null, null, callable, operationTimeout);
+ ars.waitUntilDone();
+ if (ars.hasError()) {
+ throw ars.getErrors();
+ }
}
/**
@@ -768,21 +774,30 @@ public class HTable implements Table {
final byte [] qualifier, final CompareOp compareOp, final byte [] value,
final Delete delete)
throws IOException {
- RegionServerCallable<Boolean> callable =
- new RegionServerCallable<Boolean>(this.connection, this.rpcControllerFactory,
- getName(), row) {
+ CancellableRegionServerCallable<SingleResponse> callable =
+ new CancellableRegionServerCallable<SingleResponse>(
+ this.connection, getName(), row, this.rpcControllerFactory) {
@Override
- protected Boolean rpcCall() throws Exception {
+ protected SingleResponse rpcCall() throws Exception {
CompareType compareType = CompareType.valueOf(compareOp.name());
MutateRequest request = RequestConverter.buildMutateRequest(
getLocation().getRegionInfo().getRegionName(), row, family, qualifier,
new BinaryComparator(value), compareType, delete);
MutateResponse response = getStub().mutate(getRpcController(), request);
- return Boolean.valueOf(response.getProcessed());
+ return ResponseConverter.getResult(request, response, getRpcControllerCellScanner());
}
};
- return rpcCallerFactory.<Boolean> newCaller(this.writeRpcTimeout).callWithRetries(callable,
- this.operationTimeout);
+ List<Row> rows = new ArrayList<Row>();
+ rows.add(delete);
+
+ Object[] results = new Object[1];
+ AsyncRequestFuture ars = multiAp.submitAll(pool, tableName, rows,
+ null, results, callable, operationTimeout);
+ ars.waitUntilDone();
+ if (ars.hasError()) {
+ throw ars.getErrors();
+ }
+ return ((SingleResponse.Entry)results[0]).isProcessed();
}
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/831fb3cc/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiResponse.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiResponse.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiResponse.java
index 79a9ed3..18376f4 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiResponse.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiResponse.java
@@ -31,7 +31,7 @@ import org.apache.hadoop.hbase.util.Bytes;
* A container for Result objects, grouped by regionName.
*/
@InterfaceAudience.Private
-public class MultiResponse {
+public class MultiResponse extends AbstractResponse {
// map of regionName to map of Results by the original index for that Result
private Map<byte[], RegionResult> results = new TreeMap<>(Bytes.BYTES_COMPARATOR);
@@ -101,6 +101,11 @@ public class MultiResponse {
return this.results;
}
+ @Override
+ public ResponseType type() {
+ return ResponseType.MULTI;
+ }
+
static class RegionResult{
Map<Integer, Object> result = new HashMap<>();
ClientProtos.RegionLoadStats stat;
http://git-wip-us.apache.org/repos/asf/hbase/blob/831fb3cc/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SingleResponse.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SingleResponse.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SingleResponse.java
new file mode 100644
index 0000000..68897b5
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SingleResponse.java
@@ -0,0 +1,65 @@
+/*
+ *
+ * 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.client;
+
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+
+/**
+ * Class for single action response
+ */
+@InterfaceAudience.Private
+public class SingleResponse extends AbstractResponse {
+ private Entry entry = null;
+
+ @InterfaceAudience.Private
+ public static class Entry {
+ private Result result = null;
+ private boolean processed = false;
+
+ public Result getResult() {
+ return result;
+ }
+
+ public void setResult(Result result) {
+ this.result = result;
+ }
+
+ public boolean isProcessed() {
+ return processed;
+ }
+
+ public void setProcessed(boolean processed) {
+ this.processed = processed;
+ }
+
+ }
+
+ public Entry getEntry() {
+ return entry;
+ }
+
+ public void setEntry(Entry entry) {
+ this.entry = entry;
+ }
+ @Override
+ public ResponseType type() {
+ return ResponseType.SINGLE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/831fb3cc/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ResponseConverter.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ResponseConverter.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ResponseConverter.java
index 76b4ccf..e5deabd 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ResponseConverter.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ResponseConverter.java
@@ -33,6 +33,7 @@ import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.SingleResponse;
import org.apache.hadoop.hbase.ipc.ServerRpcController;
import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.GetUserPermissionsResponse;
import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CloseRegionResponse;
@@ -149,6 +150,19 @@ public final class ResponseConverter {
return results;
}
+
+ public static SingleResponse getResult(final ClientProtos.MutateRequest request,
+ final ClientProtos.MutateResponse response,
+ final CellScanner cells)
+ throws IOException {
+ SingleResponse singleResponse = new SingleResponse();
+ SingleResponse.Entry entry = new SingleResponse.Entry();
+ entry.setResult(ProtobufUtil.toResult(response.getResult(), cells));
+ entry.setProcessed(response.getProcessed());
+ singleResponse.setEntry(entry);
+ return singleResponse;
+ }
+
/**
* Wrap a throwable to an action result.
*
http://git-wip-us.apache.org/repos/asf/hbase/blob/831fb3cc/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
index e7366a9..54552d9 100644
--- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
+++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
@@ -218,7 +218,7 @@ public class TestAsyncProcess {
// Do nothing for avoiding the NPE if we test the ClientBackofPolicy.
}
@Override
- protected RpcRetryingCaller<MultiResponse> createCaller(
+ protected RpcRetryingCaller<AbstractResponse> createCaller(
CancellableRegionServerCallable callable) {
callsCt.incrementAndGet();
MultiServerCallable callable1 = (MultiServerCallable) callable;
@@ -234,9 +234,9 @@ public class TestAsyncProcess {
}
});
- return new RpcRetryingCallerImpl<MultiResponse>(100, 10, 9) {
+ return new RpcRetryingCallerImpl<AbstractResponse>(100, 10, 9) {
@Override
- public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
+ public AbstractResponse callWithoutRetries(RetryingCallable<AbstractResponse> callable,
int callTimeout)
throws IOException, RuntimeException {
try {
@@ -252,7 +252,7 @@ public class TestAsyncProcess {
}
}
- static class CallerWithFailure extends RpcRetryingCallerImpl<MultiResponse>{
+ static class CallerWithFailure extends RpcRetryingCallerImpl<AbstractResponse>{
private final IOException e;
@@ -262,7 +262,7 @@ public class TestAsyncProcess {
}
@Override
- public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
+ public AbstractResponse callWithoutRetries(RetryingCallable<AbstractResponse> callable,
int callTimeout)
throws IOException, RuntimeException {
throw e;
@@ -281,7 +281,7 @@ public class TestAsyncProcess {
}
@Override
- protected RpcRetryingCaller<MultiResponse> createCaller(
+ protected RpcRetryingCaller<AbstractResponse> createCaller(
CancellableRegionServerCallable callable) {
callsCt.incrementAndGet();
return new CallerWithFailure(ioe);
@@ -332,7 +332,7 @@ public class TestAsyncProcess {
}
@Override
- protected RpcRetryingCaller<MultiResponse> createCaller(
+ protected RpcRetryingCaller<AbstractResponse> createCaller(
CancellableRegionServerCallable payloadCallable) {
MultiServerCallable<Row> callable = (MultiServerCallable) payloadCallable;
final MultiResponse mr = createMultiResponse(
@@ -362,9 +362,9 @@ public class TestAsyncProcess {
replicaCalls.incrementAndGet();
}
- return new RpcRetryingCallerImpl<MultiResponse>(100, 10, 9) {
+ return new RpcRetryingCallerImpl<AbstractResponse>(100, 10, 9) {
@Override
- public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
+ public MultiResponse callWithoutRetries(RetryingCallable<AbstractResponse> callable,
int callTimeout)
throws IOException, RuntimeException {
long sleep = -1;
http://git-wip-us.apache.org/repos/asf/hbase/blob/831fb3cc/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java
index bc94b02..f465625 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java
@@ -86,6 +86,7 @@ import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
import org.apache.hadoop.hbase.master.HMaster;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto;
import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType;
import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MultiRowMutationService;
@@ -1858,6 +1859,33 @@ public class TestFromClientSide {
}
@Test
+ public void testDeleteWithFailed() throws Exception {
+ TableName TABLE = TableName.valueOf("testDeleteWithFailed");
+
+ byte [][] ROWS = makeNAscii(ROW, 6);
+ byte [][] FAMILIES = makeNAscii(FAMILY, 3);
+ byte [][] VALUES = makeN(VALUE, 5);
+ long [] ts = {1000, 2000, 3000, 4000, 5000};
+
+ Table ht = TEST_UTIL.createTable(TABLE, FAMILIES, 3);
+
+ Put put = new Put(ROW);
+ put.addColumn(FAMILIES[0], QUALIFIER, ts[0], VALUES[0]);
+ ht.put(put);
+
+ // delete wrong family
+ Delete delete = new Delete(ROW);
+ delete.addFamily(FAMILIES[1], ts[0]);
+ ht.delete(delete);
+
+ Get get = new Get(ROW);
+ get.addFamily(FAMILIES[0]);
+ get.setMaxVersions(Integer.MAX_VALUE);
+ Result result = ht.get(get);
+ assertTrue(Bytes.equals(result.getValue(FAMILIES[0], QUALIFIER), VALUES[0]));
+ }
+
+ @Test
public void testDeletes() throws Exception {
TableName TABLE = TableName.valueOf("testDeletes");
@@ -4623,6 +4651,24 @@ public class TestFromClientSide {
}
@Test
+ public void testCheckAndDelete() throws IOException {
+ final byte [] value1 = Bytes.toBytes("aaaa");
+
+ Table table = TEST_UTIL.createTable(TableName.valueOf("testCheckAndDelete"),
+ FAMILY);
+
+ Put put = new Put(ROW);
+ put.addColumn(FAMILY, QUALIFIER, value1);
+ table.put(put);
+
+ Delete delete = new Delete(ROW);
+ delete.addColumns(FAMILY, QUALIFIER);
+
+ boolean ok = table.checkAndDelete(ROW, FAMILY, QUALIFIER, value1, delete);
+ assertEquals(ok, true);
+ }
+
+ @Test
public void testCheckAndDeleteWithCompareOp() throws IOException {
final byte [] value1 = Bytes.toBytes("aaaa");
final byte [] value2 = Bytes.toBytes("bbbb");
[3/7] hbase git commit: HBASE-16229 Cleaning up size and heapSize
calculation.
Posted by an...@apache.org.
http://git-wip-us.apache.org/repos/asf/hbase/blob/2ab33846/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
index a6c7912..74826b0 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestWalAndCompactingMemStoreFlush.java
@@ -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();
- }
}
[6/7] hbase git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/hbase
Posted by an...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/hbase
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/e6f8f6db
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/e6f8f6db
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/e6f8f6db
Branch: refs/heads/master
Commit: e6f8f6dbd0f7e801fd0fbafbe8cb35492bf08285
Parents: 156a8b2 1d6c90b
Author: anoopsamjohn <an...@gmail.com>
Authored: Thu Sep 15 18:07:12 2016 +0530
Committer: anoopsamjohn <an...@gmail.com>
Committed: Thu Sep 15 18:07:12 2016 +0530
----------------------------------------------------------------------
.../org/apache/hadoop/hbase/client/Admin.java | 5 +-
.../hadoop/hbase/client/AsyncProcess.java | 4 +-
.../apache/hadoop/hbase/client/HBaseAdmin.java | 6 +-
.../hadoop/hbase/ipc/AbstractRpcClient.java | 22 ++++
.../hbase/ipc/ServerTooBusyException.java | 38 ++++++
.../org/apache/hadoop/hbase/HConstants.java | 12 ++
.../java/org/apache/hadoop/hbase/KeyValue.java | 15 ++-
.../src/main/resources/hbase-default.xml | 16 ++-
.../AbstractStateMachineNamespaceProcedure.java | 70 +++++++++++
.../AbstractStateMachineTableProcedure.java | 111 +++++++++++++++++
.../procedure/AddColumnFamilyProcedure.java | 38 ++----
.../procedure/CloneSnapshotProcedure.java | 30 +----
.../procedure/CreateNamespaceProcedure.java | 28 +----
.../master/procedure/CreateTableProcedure.java | 43 ++-----
.../procedure/DeleteColumnFamilyProcedure.java | 38 ++----
.../procedure/DeleteNamespaceProcedure.java | 34 +-----
.../master/procedure/DeleteTableProcedure.java | 49 ++------
.../master/procedure/DisableTableProcedure.java | 55 ++-------
.../DispatchMergingRegionsProcedure.java | 16 +--
.../master/procedure/EnableTableProcedure.java | 55 ++-------
.../procedure/MasterDDLOperationHelper.java | 16 ---
.../procedure/ModifyColumnFamilyProcedure.java | 38 ++----
.../procedure/ModifyNamespaceProcedure.java | 34 +-----
.../master/procedure/ModifyTableProcedure.java | 44 ++-----
.../procedure/RestoreSnapshotProcedure.java | 27 +----
.../procedure/TruncateTableProcedure.java | 36 ++----
.../org/apache/hadoop/hbase/client/TestHCM.java | 119 ++++++++++++++++++-
.../TestScannerHeartbeatMessages.java | 105 +++++++---------
hbase-shell/src/main/ruby/hbase/security.rb | 9 +-
hbase-shell/src/main/ruby/hbase/table.rb | 81 +++++++++----
.../src/main/ruby/shell/commands/deleteall.rb | 17 ++-
hbase-shell/src/test/ruby/hbase/table_test.rb | 12 ++
src/main/asciidoc/_chapters/developer.adoc | 60 ++++++----
33 files changed, 677 insertions(+), 606 deletions(-)
----------------------------------------------------------------------