HBASE-1252 Make atomic increment perform a binary increment

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

CHANGES.txt

@@ -87,6 +87,8 @@ Release 0.20.0 - Unreleased
    HBASE-1240  Would be nice if RowResult could be comparable
                (Erik Holstad via Stack)
    HBASE-803   Atomic increment operations (Ryan Rawson and Jon Gray via Stack)
+   HBASE-1252  Make atomic increment perform a binary increment
+               (Jonathan Gray via Stack)
 
 Release 0.19.0 - 01/21/2009
   INCOMPATIBLE CHANGES

src/java/org/apache/hadoop/hbase/client/HTable.java

@@ -1481,7 +1481,7 @@ public class HTable {
   }
   
   public long incrementColumnValue(final byte [] row, final byte [] column,
-      final int amount) throws IOException {
+      final long amount) throws IOException {
     return connection.getRegionServerWithRetries(
         new ServerCallable<Long>(connection, tableName, row) {
           public Long call() throws IOException {

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

@@ -2598,7 +2598,6 @@ public class HRegion implements HConstants {
       HStoreKey hsk = new HStoreKey(row, column);
       long ts = System.currentTimeMillis();
       byte [] value = null;
-      long newval; // the new value.
 
       Store store = getStore(column);
 
@@ -2633,22 +2632,22 @@ public class HRegion implements HConstants {
       if (value == null) {
         // Doesn't exist
         LOG.debug("Creating new counter value for " + Bytes.toString(row) + "/"+ Bytes.toString(column));
-        newval = amount;
+        value = Bytes.toBytes(amount);
       } else {
-        newval = incrementBytes(value, amount);
+        value = incrementBytes(value, amount);
       }
 
       BatchUpdate b = new BatchUpdate(row, ts);
-      b.put(column, Bytes.toBytes(newval));
+      b.put(column, value);
       batchUpdate(b, lid, true);
-      return newval;
+      return Bytes.toLong(value);
     } finally {
       splitsAndClosesLock.readLock().unlock();
       releaseRowLock(lid);
     }
   }
 
-  private long incrementBytes(byte[] value, long amount) throws IOException {
+  private byte [] incrementBytes(byte[] value, long amount) throws IOException {
     // Hopefully this doesn't happen too often.
     if (value.length < Bytes.SIZEOF_LONG) {
       byte [] newvalue = new byte[Bytes.SIZEOF_LONG];
@@ -2657,8 +2656,22 @@ public class HRegion implements HConstants {
     } else if (value.length > Bytes.SIZEOF_LONG) {
       throw new DoNotRetryIOException("Increment Bytes - value too big: " + value.length);
     }
-    long v = Bytes.toLong(value);
-    v += amount;
-    return v;
+    return binaryIncrement(value, amount);
+  }
+  
+  private byte [] binaryIncrement(byte [] value, long amount) {
+    for(int i=0;i<value.length;i++) {
+      int cur = (int)(amount >> (8 * i)) % 256;
+      int val = (int)(value[value.length-i-1] & 0xff);
+      int total = cur + val;
+      if(total > 255) {
+        amount += ((long)256 << (8 * i));
+        total %= 256;
+      }
+      value[value.length-i-1] = (byte)total;
+      amount = (amount >> (8 * (i + 1))) << (8 * (i + 1));
+      if(amount == 0) return value;
+    }
+    return value;
   }
 }