HBASE-11541 Wrong result when scaning meta with startRow (Liu Shaoqui)

hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java

@@ -1663,20 +1663,6 @@ public class KeyValue implements Cell, HeapSize, Cloneable {
     return Bytes.add(family, COLUMN_FAMILY_DELIM_ARRAY, qualifier);
   }
 
-  /**
-   * This function is only used in Meta key comparisons so its error message
-   * is specific for meta key errors.
-   */
-  static int getRequiredDelimiterInReverse(final byte [] b,
-      final int offset, final int length, final int delimiter) {
-    int index = getDelimiterInReverse(b, offset, length, delimiter);
-    if (index < 0) {
-      throw new IllegalArgumentException("hbase:meta key must have two '" + (char)delimiter + "' "
-        + "delimiters and have the following format: '<table>,<key>,<etc>'");
-    }
-    return index;
-  }
-
   /**
    * @param b
    * @param delimiter
@@ -1749,35 +1735,44 @@ public class KeyValue implements Cell, HeapSize, Cloneable {
           HConstants.DELIMITER);
       int rightDelimiter = getDelimiter(right, roffset, rlength,
           HConstants.DELIMITER);
-      if (leftDelimiter < 0 && rightDelimiter >= 0) {
-        // Nothing between hbase:meta and regionid.  Its first key.
-        return -1;
-      } else if (rightDelimiter < 0 && leftDelimiter >= 0) {
-        return 1;
-      } else if (leftDelimiter < 0 && rightDelimiter < 0) {
-        return 0;
-      }
       // Compare up to the delimiter
-      int result = Bytes.compareTo(left, loffset, leftDelimiter - loffset,
+      int lpart = (leftDelimiter < 0 ? llength :leftDelimiter - loffset);
-          right, roffset, rightDelimiter - roffset);
+      int rpart = (rightDelimiter < 0 ? rlength :rightDelimiter - roffset);
+      int result = Bytes.compareTo(left, loffset, lpart, right, roffset, rpart);
       if (result != 0) {
         return result;
+      } else {
+        if (leftDelimiter < 0 && rightDelimiter >= 0) {
+          return -1;
+        } else if (rightDelimiter < 0 && leftDelimiter >= 0) {
+          return 1;
+        } else if (leftDelimiter < 0 && rightDelimiter < 0) {
+          return 0;
+        }
       }
       // Compare middle bit of the row.
       // Move past delimiter
       leftDelimiter++;
       rightDelimiter++;
-      int leftFarDelimiter = getRequiredDelimiterInReverse(left, leftDelimiter,
+      int leftFarDelimiter = getDelimiterInReverse(left, leftDelimiter,
           llength - (leftDelimiter - loffset), HConstants.DELIMITER);
-      int rightFarDelimiter = getRequiredDelimiterInReverse(right,
+      int rightFarDelimiter = getDelimiterInReverse(right,
           rightDelimiter, rlength - (rightDelimiter - roffset),
           HConstants.DELIMITER);
       // Now compare middlesection of row.
-      result = super.compareRows(left, leftDelimiter,
+      lpart = (leftFarDelimiter < 0 ? llength + loffset: leftFarDelimiter) - leftDelimiter;
-          leftFarDelimiter - leftDelimiter, right, rightDelimiter,
+      rpart = (rightFarDelimiter < 0 ? rlength + roffset: rightFarDelimiter)- rightDelimiter;
-          rightFarDelimiter - rightDelimiter);
+      result = super.compareRows(left, leftDelimiter, lpart, right, rightDelimiter, rpart);
       if (result != 0) {
         return result;
+      }  else {
+        if (leftDelimiter < 0 && rightDelimiter >= 0) {
+          return -1;
+        } else if (rightDelimiter < 0 && leftDelimiter >= 0) {
+          return 1;
+        } else if (leftDelimiter < 0 && rightDelimiter < 0) {
+          return 0;
+        }
       }
       // Compare last part of row, the rowid.
       leftFarDelimiter++;

hbase-common/src/test/java/org/apache/hadoop/hbase/TestKeyValue.java

@@ -150,23 +150,6 @@ public class TestKeyValue extends TestCase {
     metacomparisons(new KeyValue.MetaComparator());
   }
 
-  public void testBadMetaCompareSingleDelim() {
-    MetaComparator c = new KeyValue.MetaComparator();
-    long now = System.currentTimeMillis();
-    // meta keys values are not quite right.  A users can enter illegal values
-    // from shell when scanning meta.
-    KeyValue a = new KeyValue(Bytes.toBytes("table,a1"), now);
-    KeyValue b = new KeyValue(Bytes.toBytes("table,a2"), now);
-    try {
-      c.compare(a, b);
-    } catch (IllegalArgumentException iae) {
-      assertEquals("hbase:meta key must have two ',' delimiters and have the following" +
-      		" format: '<table>,<key>,<etc>'", iae.getMessage());
-      return;
-    }
-    fail("Expected IllegalArgumentException");
-  }
-
   public void testMetaComparatorTableKeysWithCommaOk() {
     MetaComparator c = new KeyValue.MetaComparator();
     long now = System.currentTimeMillis();
@@ -589,4 +572,49 @@ public class TestKeyValue extends TestCase {
     Bytes.equals(next.getValue(), metaValue2);
     assertFalse(tagItr.hasNext());
   }
+  
+  public void testMetaKeyComparator() {
+    MetaComparator c = new KeyValue.MetaComparator();
+    long now = System.currentTimeMillis();
+
+    KeyValue a = new KeyValue(Bytes.toBytes("table1"), now);
+    KeyValue b = new KeyValue(Bytes.toBytes("table2"), now);
+    assertTrue(c.compare(a, b) < 0);
+    
+    a = new KeyValue(Bytes.toBytes("table1,111"), now);
+    b = new KeyValue(Bytes.toBytes("table2"), now);
+    assertTrue(c.compare(a, b) < 0);
+    
+    a = new KeyValue(Bytes.toBytes("table1"), now);
+    b = new KeyValue(Bytes.toBytes("table2,111"), now);
+    assertTrue(c.compare(a, b) < 0);
+    
+    a = new KeyValue(Bytes.toBytes("table,111"), now);
+    b = new KeyValue(Bytes.toBytes("table,2222"), now);
+    assertTrue(c.compare(a, b) < 0);
+    
+    a = new KeyValue(Bytes.toBytes("table,111,aaaa"), now);
+    b = new KeyValue(Bytes.toBytes("table,2222"), now);
+    assertTrue(c.compare(a, b) < 0);
+    
+    a = new KeyValue(Bytes.toBytes("table,111"), now);
+    b = new KeyValue(Bytes.toBytes("table,2222.bbb"), now);
+    assertTrue(c.compare(a, b) < 0);
+
+    a = new KeyValue(Bytes.toBytes("table,,aaaa"), now);
+    b = new KeyValue(Bytes.toBytes("table,111,bbb"), now);
+    assertTrue(c.compare(a, b) < 0);
+    
+    a = new KeyValue(Bytes.toBytes("table,111,aaaa"), now);
+    b = new KeyValue(Bytes.toBytes("table,111,bbb"), now);
+    assertTrue(c.compare(a, b) < 0);
+
+    a = new KeyValue(Bytes.toBytes("table,111,xxxx"), now);
+    b = new KeyValue(Bytes.toBytes("table,111,222,bbb"), now);
+    assertTrue(c.compare(a, b) < 0);
+    
+    a = new KeyValue(Bytes.toBytes("table,111,11,xxx"), now);
+    b = new KeyValue(Bytes.toBytes("table,111,222,bbb"), now);
+    assertTrue(c.compare(a, b) < 0);
+  }
 }