HBASE-712 midKey found compacting is the first, not necessarily the optimal

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

CHANGES.txt

@@ -163,6 +163,7 @@ Trunk (unreleased changes)
    HBASE-527   RegexpRowFilter does not work when there are columns from 
                multiple families (Clint Morgan via Jim Kellerman)
    HBASE-534   Double-assignment at SPLIT-time
+   HBASE-712   midKey found compacting is the first, not necessarily the optimal
    
   IMPROVEMENTS
    HBASE-559   MR example job to count table rows

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

@@ -876,10 +876,12 @@ public class HRegion implements HConstants {
       LOG.info("starting compaction on region " + this);
       long startTime = System.currentTimeMillis();
       doRegionCompactionPrep();
+      long maxSize = -1;
       for (HStore store: stores.values()) {
-        final byte [] key = store.compact(force);
+        final HStore.StoreSize size = store.compact(force);
-        if (key != null && midKey == null) {
+        if (size != null && size.getSize() > maxSize) {
-          midKey = key;
+          maxSize = size.getSize();
+          midKey = size.getKey();
         }
       }
       doRegionCompactionCleanup();

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

@@ -854,7 +854,7 @@ public class HStore implements HConstants {
    * @return mid key if a split is needed, null otherwise
    * @throws IOException
    */
-  byte [] compact(final boolean force) throws IOException {
+  StoreSize compact(final boolean force) throws IOException {
     synchronized (compactLock) {
       long maxId = -1;
       List<HStoreFile> filesToCompact = null;
@@ -1811,9 +1811,9 @@ public class HStore implements HConstants {
   /**
    * Determines if HStore can be split
    * 
-   * @return midKey if store can be split, null otherwise
+   * @return a StoreSize if store can be split, null otherwise
    */
-  byte [] checkSplit() {
+  StoreSize checkSplit() {
     if (this.storefiles.size() <= 0) {
       return null;
     }
@@ -1865,7 +1865,7 @@ public class HStore implements HConstants {
             Bytes.equals(mk.getRow(), lastKey.getRow())) {
           return null;
         }
-        return mk.getRow();
+        return new StoreSize(maxSize, mk.getRow());
       }
     } catch(IOException e) {
       LOG.warn("Failed getting store size for " + this.storeNameStr, e);
@@ -1931,4 +1931,22 @@ public class HStore implements HConstants {
       return copy;
     }
   }
+  
+  class StoreSize {
+    private final long size;
+    private final byte[] key;
+    StoreSize(long size, byte[] key) {
+      this.size = size;
+      this.key = new byte[key.length];
+      System.arraycopy(key, 0, this.key, 0, key.length);
+    }
+    /* @return the size */
+    long getSize() {
+      return size;
+    }
+    /* @return the key */
+    byte[] getKey() {
+      return key;
+    }
+  }
 }