HADOOP-2493 hbase will split on row when the start and end row

is the same cause data loss


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

CHANGES.txt

@@ -131,6 +131,8 @@ Trunk (unreleased changes)
                duration of compaction. 
    HADOOP-2592 Scanning, a region can let out a row that its not supposed
                to have
+   HADOOP-2493 hbase will split on row when the start and end row is the
+               same cause data loss (Bryan Duxbury via Stack)
    
   IMPROVEMENTS
    HADOOP-2401 Add convenience put method that takes writable

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

@@ -654,7 +654,8 @@ public class HRegion implements HConstants {
    */
   boolean needsSplit(Text midKey) {
     HStore.HStoreSize biggest = largestHStore(midKey);
-    if (biggest == null) {
+    if (biggest == null || midKey.getLength() == 0 || 
+      (midKey.equals(getStartKey()) && midKey.equals(getEndKey())) ) {
       return false;
     }
     long triggerSize = this.desiredMaxFileSize + (this.desiredMaxFileSize / 2);

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

@@ -2020,9 +2020,28 @@ public class HStore implements HConstants {
         }
       }
       MapFile.Reader r = this.readers.get(mapIndex);
-      WritableComparable midkey = r.midKey();
+      
+      // seek back to the beginning of mapfile
+      r.reset();
+      
+      // get the first and last keys
+      HStoreKey firstKey = new HStoreKey();
+      HStoreKey lastKey = new HStoreKey();
+      Writable value = new ImmutableBytesWritable();
+      r.next((WritableComparable)firstKey, value);
+      r.finalKey((WritableComparable)lastKey);
+      
+      // get the midkey
+      HStoreKey midkey = (HStoreKey)r.midKey();
+
       if (midkey != null) {
         midKey.set(((HStoreKey)midkey).getRow());
+        // if the midkey is the same as the first and last keys, then we cannot
+        // (ever) split this region. 
+        if (midkey.getRow().equals(firstKey.getRow()) && 
+          midkey.getRow().equals(lastKey.getRow())) {
+          return new HStoreSize(aggregateSize, maxSize, false);
+        } 
       }
     } catch(IOException e) {
       LOG.warn("Failed getting store size for " + this.storeName, e);

src/java/org/apache/hadoop/hbase/HStoreFile.java

@@ -770,11 +770,18 @@ public class HStoreFile implements HConstants {
     }
 
     /** {@inheritDoc} */
-    @SuppressWarnings({ "unused"})
     @Override
     public synchronized void finalKey(WritableComparable key)
     throws IOException {
-      throw new UnsupportedOperationException("Unsupported");
+      if (top) {
+        checkKey(key);
+        super.finalKey(key); 
+      } else {
+        reset();
+        Writable value = new ImmutableBytesWritable();
+        
+        key = super.getClosest(midkey, value, true);
+      }
     }
 
     /** {@inheritDoc} */