HBASE-4476 Compactions must fail if column tracker gets columns out of order

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

CHANGES.txt

@@ -322,6 +322,8 @@ Release 0.92.0 - Unreleased
    HBASE-4212  TestMasterFailover fails occasionally (Gao Jinchao)
    HBASE-4412  No need to retry scan operation on the same server in case of
                RegionServerStoppedException (Ming Ma)
+   HBASE-4476  Compactions must fail if column tracker gets columns out of order
+               (Mikhail Bautin)
 
   TESTS
    HBASE-4450  test for number of blocks read: to serve as baseline for expected

src/main/java/org/apache/hadoop/hbase/regionserver/ColumnTracker.java

@@ -19,6 +19,8 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
+import java.io.IOException;
+
 import org.apache.hadoop.hbase.regionserver.ScanQueryMatcher.MatchCode;
 
 /**
@@ -46,9 +48,11 @@ public interface ColumnTracker {
    * @param length
    * @param ttl The timeToLive to enforce.
    * @return The match code instance.
+   * @throws IOException in case there is an internal consistency problem
+   *      caused by a data corruption.
    */
   public ScanQueryMatcher.MatchCode checkColumn(byte [] bytes, int offset,
-      int length, long ttl);
+      int length, long ttl) throws IOException;
 
   /**
    * Updates internal variables in between files

src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java

@@ -28,6 +28,7 @@ import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
 import org.apache.hadoop.hbase.io.TimeRange;
 import org.apache.hadoop.hbase.util.Bytes;
 
+import java.io.IOException;
 import java.util.NavigableSet;
 
 /**
@@ -122,8 +123,10 @@ public class ScanQueryMatcher {
    *
    * @param kv KeyValue to check
    * @return The match code instance.
+   * @throws IOException in case there is an internal consistency problem
+   *      caused by a data corruption.
    */
-  public MatchCode match(KeyValue kv) {
+  public MatchCode match(KeyValue kv) throws IOException {
     if (filter != null && filter.filterAllRemaining()) {
       return MatchCode.DONE_SCAN;
     }

src/main/java/org/apache/hadoop/hbase/regionserver/ScanWildcardColumnTracker.java

@@ -20,6 +20,8 @@
 
 package org.apache.hadoop.hbase.regionserver;
 
+import java.io.IOException;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.HConstants;
@@ -68,7 +70,7 @@ public class ScanWildcardColumnTracker implements ColumnTracker {
    */
   @Override
   public MatchCode checkColumn(byte[] bytes, int offset, int length,
-      long timestamp) {
+      long timestamp) throws IOException {
     if (columnBuffer == null) {
       // first iteration.
       resetBuffer(bytes, offset, length);
@@ -94,16 +96,13 @@ public class ScanWildcardColumnTracker implements ColumnTracker {
     }
 
     // new col < oldcol
-    // if (cmp < 0) {
     // WARNING: This means that very likely an edit for some other family
-    // was incorrectly stored into the store for this one. Continue, but
-    // complain.
-    LOG.error("ScanWildcardColumnTracker.checkColumn ran " +
-        "into a column actually smaller than the previous column: " +
-      Bytes.toStringBinary(bytes, offset, length));
-    // switched columns
-    resetBuffer(bytes, offset, length);
-    return checkVersion(++currentCount, timestamp);
+    // was incorrectly stored into the store for this one. Throw an exception,
+    // because this might lead to data corruption.
+    throw new IOException(
+        "ScanWildcardColumnTracker.checkColumn ran into a column actually " +
+        "smaller than the previous column: " +
+        Bytes.toStringBinary(bytes, offset, length));
   }
 
   private void resetBuffer(byte[] bytes, int offset, int length) {

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

@@ -1803,4 +1803,9 @@ public class Store implements HeapSize {
   public long heapSize() {
     return DEEP_OVERHEAD + this.memstore.heapSize();
   }
+
+  public KeyValue.KVComparator getComparator() {
+    return comparator;
+  }
+
 }

src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java

@@ -255,12 +255,25 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
     }
 
     KeyValue kv;
+    KeyValue prevKV = null;
     List<KeyValue> results = new ArrayList<KeyValue>();
+
+    // Only do a sanity-check if store and comparator are available.
+    KeyValue.KVComparator comparator =
+        store != null ? store.getComparator() : null;
+
     LOOP: while((kv = this.heap.peek()) != null) {
       // kv is no longer immutable due to KeyOnlyFilter! use copy for safety
       KeyValue copyKv = kv.shallowCopy();
+      // Check that the heap gives us KVs in an increasing order.
+      if (prevKV != null && comparator != null
+          && comparator.compare(prevKV, kv) > 0) {
+        throw new IOException("Key " + prevKV + " followed by a " +
+            "smaller key " + kv + " in cf " + store);
+      }
+      prevKV = copyKv;
       ScanQueryMatcher.MatchCode qcode = matcher.match(copyKv);
-      //DebugPrint.println("SS peek kv = " + kv + " with qcode = " + qcode);
+
       switch(qcode) {
         case INCLUDE:
         case INCLUDE_AND_SEEK_NEXT_ROW:

src/test/java/org/apache/hadoop/hbase/regionserver/TestExplicitColumnTracker.java

@@ -20,6 +20,7 @@
 
 package org.apache.hadoop.hbase.regionserver;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.TreeSet;
@@ -42,7 +43,7 @@ public class TestExplicitColumnTracker extends HBaseTestCase {
   private void runTest(int maxVersions,
                        TreeSet<byte[]> trackColumns,
                        List<byte[]> scannerColumns,
-                       List<MatchCode> expected) {
+                       List<MatchCode> expected) throws IOException {
     ColumnTracker exp = new ExplicitColumnTracker(
       trackColumns, 0, maxVersions, Long.MAX_VALUE);
 
@@ -66,7 +67,7 @@ public class TestExplicitColumnTracker extends HBaseTestCase {
     }
   }
 
-  public void testGet_SingleVersion(){
+  public void testGet_SingleVersion() throws IOException{
     if(PRINT){
       System.out.println("SingleVersion");
     }
@@ -95,7 +96,7 @@ public class TestExplicitColumnTracker extends HBaseTestCase {
     runTest(maxVersions, columns, scanner, expected);
   }
 
-  public void testGet_MultiVersion(){
+  public void testGet_MultiVersion() throws IOException{
     if(PRINT){
       System.out.println("\nMultiVersion");
     }
@@ -154,7 +155,7 @@ public class TestExplicitColumnTracker extends HBaseTestCase {
   /**
    * hbase-2259
    */
-  public void testStackOverflow(){
+  public void testStackOverflow() throws IOException{
     int maxVersions = 1;
     TreeSet<byte[]> columns = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
     for (int i = 0; i < 100000; i++) {
@@ -178,7 +179,7 @@ public class TestExplicitColumnTracker extends HBaseTestCase {
   /**
    * Regression test for HBASE-2545
    */
-  public void testInfiniteLoop() {
+  public void testInfiniteLoop() throws IOException {
     TreeSet<byte[]> columns = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
     columns.addAll(Arrays.asList(new byte[][] {
       col2, col3, col5 }));

src/test/java/org/apache/hadoop/hbase/regionserver/TestScanWildcardColumnTracker.java

@@ -20,6 +20,7 @@
 
 package org.apache.hadoop.hbase.regionserver;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -31,7 +32,7 @@ public class TestScanWildcardColumnTracker extends HBaseTestCase {
 
   final static int VERSIONS = 2;
 
-  public void testCheckColumn_Ok() {
+  public void testCheckColumn_Ok() throws IOException {
     ScanWildcardColumnTracker tracker =
       new ScanWildcardColumnTracker(0, VERSIONS, Long.MAX_VALUE);
 
@@ -63,7 +64,7 @@ public class TestScanWildcardColumnTracker extends HBaseTestCase {
     }
   }
 
-  public void testCheckColumn_EnforceVersions() {
+  public void testCheckColumn_EnforceVersions() throws IOException {
     ScanWildcardColumnTracker tracker =
       new ScanWildcardColumnTracker(0, VERSIONS, Long.MAX_VALUE);