HBASE-1547 atomicIncrement doesnt increase hregion.memcacheSize

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hbase/trunk@786693 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -209,6 +209,7 @@ Release 0.20.0 - Unreleased
    HBASE-1387  Before release verify all object sizes using Ryans' instrumented
                JVM trick (Erik Holstad via Stack)
    HBASE-1545  atomicIncrements creating new values with Long.MAX_VALUE
+   HBASE-1547  atomicIncrement doesnt increase hregion.memcacheSize
 
   IMPROVEMENTS
    HBASE-1089  Add count of regions on filesystem to master UI; add percentage

src/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -2276,15 +2276,28 @@ public class HRegion implements HConstants { // , Writable{
   throws IOException {
     checkRow(row);
 
+    boolean flush = false;
     // Lock row
     Integer lid = obtainRowLock(row);
     long result = 0L;
     try {
       Store store = stores.get(family);
-      result = store.incrementColumnValue(row, family, qualifier, amount);
+
+      Store.ValueAndSize vas =
+          store.incrementColumnValue(row, family, qualifier, amount);
+
+      result = vas.value;
+      long size = this.memcacheSize.addAndGet(vas.sizeAdded);
+      flush = isFlushSize(size);
     } finally {
       releaseRowLock(lid);
     }
+
+    if (flush) {
+      // Request a cache flush.  Do it outside update lock.
+      requestFlush();
+    }
+
     return result;
   }
     

src/java/org/apache/hadoop/hbase/regionserver/Store.java

@@ -1495,6 +1495,15 @@ public class Store implements HConstants {
     scanner.get(result);
   }
 
+  public static class ValueAndSize {
+    public long value;
+    public long sizeAdded;
+    public ValueAndSize(long value, long sizeAdded) {
+      this.value = value;
+      this.sizeAdded = sizeAdded;
+    }
+  }
+  
   /**
    * Increments the value for the given row/family/qualifier
    * @param row
@@ -1504,7 +1513,7 @@ public class Store implements HConstants {
    * @return The new value.
    * @throws IOException
    */
-  public long incrementColumnValue(byte [] row, byte [] family,
+  public ValueAndSize incrementColumnValue(byte [] row, byte [] family,
       byte [] qualifier, long amount) throws IOException {
     long value = 0;
     List<KeyValue> result = new ArrayList<KeyValue>();
@@ -1527,9 +1536,8 @@ public class Store implements HConstants {
       value = Bytes.toLong(buffer, valueOffset, Bytes.SIZEOF_LONG) + amount;
       Bytes.putBytes(buffer, valueOffset, Bytes.toBytes(value), 0, 
           Bytes.SIZEOF_LONG);
-      return value;
+      return new ValueAndSize(value, 0);
     }
-    
     // Check if we even have storefiles
     if(this.storefiles.isEmpty()) {
       return addNewKeyValue(row, family, qualifier, value, amount);
@@ -1552,14 +1560,13 @@ public class Store implements HConstants {
     return addNewKeyValue(row, family, qualifier, value, amount);
   }
   
-  private long addNewKeyValue(byte [] row, byte [] family, byte [] qualifier, 
+  private ValueAndSize addNewKeyValue(byte [] row, byte [] family, byte [] qualifier,
       long value, long amount) {
     long newValue = value + amount;
     KeyValue newKv = new KeyValue(row, family, qualifier,
         System.currentTimeMillis(),
         Bytes.toBytes(newValue));
     add(newKv);
-    return newValue;
+    return new ValueAndSize(newValue, newKv.heapSize());
   }
-  
 }