HBASE-1336 Splitting up the compare of family+column into 2 different compares

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

CHANGES.txt

@@ -110,6 +110,7 @@ Release 0.20.0 - Unreleased
    HBASE-1264  Wrong return values of comparators for ColumnValueFilter
                (Thomas Schneider via Andrew Purtell)
    HBASE-1374  NPE out of ZooKeeperWrapper.loadZooKeeperConfig
+   HBASE-1336  Splitting up the compare of family+column into 2 different compare 
 
   IMPROVEMENTS
    HBASE-1089  Add count of regions on filesystem to master UI; add percentage

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

@@ -772,13 +772,17 @@ public class KeyValue {
 
   /**
    * @param column Column minus its delimiter
+   * @param familylength Length of family in passed <code>column</code>
    * @return True if column matches.
    * @see #matchingColumn(byte[])
    */
-  public boolean matchingColumnNoDelimiter(final byte [] column) {
+  public boolean matchingColumnNoDelimiter(final byte [] column,
+      final int familylength) {
     int o = getColumnOffset();
     int l = getColumnLength(o);
-    return compareColumns(getBuffer(), o, l, column, 0, column.length) == 0;
+    int f = getFamilyLength(o);
+    return compareColumns(getBuffer(), o, l, f,
+      column, 0, column.length, familylength) == 0;
   }
 
   /**
@@ -801,15 +805,27 @@ public class KeyValue {
    * @param left
    * @param loffset
    * @param llength
+   * @param lfamilylength Offset of family delimiter in left column.
    * @param right
    * @param roffset
    * @param rlength
+   * @param rfamilylength Offset of family delimiter in right column.
    * @return
    */
   static int compareColumns(final byte [] left, final int loffset,
-      final int llength, final byte [] right, final int roffset,
+      final int llength, final int lfamilylength,
-      final int rlength) {
+      final byte [] right, final int roffset, final int rlength,
-    return Bytes.compareTo(left, loffset, llength, right, roffset, rlength);
+      final int rfamilylength) {
+    // Compare family portion first.
+    int diff = Bytes.compareTo(left, loffset, lfamilylength,
+      right, roffset, rfamilylength);
+    if (diff != 0) {
+      return diff;
+    }
+    // Compare qualifier portion
+    return Bytes.compareTo(left, loffset + lfamilylength,
+      llength - lfamilylength,
+      right, roffset + rfamilylength, rlength - rfamilylength);
   }
 
   /**
@@ -1037,11 +1053,12 @@ public class KeyValue {
     }
 
     public int compareColumns(final KeyValue left, final byte [] right,
-        final int roffset, final int rlength) {
+        final int roffset, final int rlength, final int rfamilyoffset) {
       int offset = left.getColumnOffset();
       int length = left.getColumnLength(offset);
       return getRawComparator().compareColumns(left.getBuffer(), offset, length,
-        right, roffset, rlength);
+        left.getFamilyLength(offset),
+        right, roffset, rlength, rfamilyoffset);
     }
 
     int compareColumns(final KeyValue left, final short lrowlength,
@@ -1051,9 +1068,11 @@ public class KeyValue {
       int roffset = right.getColumnOffset(rrowlength);
       int llength = left.getColumnLength(loffset, lkeylength);
       int rlength = right.getColumnLength(roffset, rkeylength);
+      int lfamilylength = left.getFamilyLength(loffset);
+      int rfamilylength = right.getFamilyLength(roffset);
       return getRawComparator().compareColumns(left.getBuffer(), loffset,
-          llength,
+          llength, lfamilylength,
-        right.getBuffer(), roffset, rlength);
+        right.getBuffer(), roffset, rlength, rfamilylength);
     }
 
     /**
@@ -1381,9 +1400,11 @@ public class KeyValue {
       return Bytes.compareTo(left, loffset, llength, right, roffset, rlength);
     }
 
-    protected int compareColumns(byte [] left, int loffset, int llength,
+    protected int compareColumns(
-        byte [] right, int roffset, int rlength) {
+        byte [] left, int loffset, int llength, final int lfamilylength,
-      return KeyValue.compareColumns(left, loffset, llength, right, roffset, rlength);
+        byte [] right, int roffset, int rlength, final int rfamilylength) {
+      return KeyValue.compareColumns(left, loffset, llength, lfamilylength,
+        right, roffset, rlength, rfamilylength);
     }
 
     int compareTimestamps(final long ltimestamp, final long rtimestamp) {

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

@@ -126,6 +126,7 @@ public abstract class HAbstractScanner implements InternalScanner {
     private Pattern columnMatcher;
     // Column without delimiter so easy compare to KeyValue column
     private byte [] col;
+    private int familylength = 0;
   
     ColumnMatcher(final byte [] col) throws IOException {
       byte [][] parse = parseColumn(col);
@@ -150,6 +151,7 @@ public abstract class HAbstractScanner implements InternalScanner {
         } else {
           this.matchType = MATCH_TYPE.SIMPLE;
           this.col = columnWithoutDelimiter;
+          this.familylength = parse[0].length;
           this.wildCardmatch = false;
         }
       } catch(Exception e) {
@@ -165,7 +167,7 @@ public abstract class HAbstractScanner implements InternalScanner {
      */
     boolean matches(final KeyValue kv) throws IOException {
       if (this.matchType == MATCH_TYPE.SIMPLE) {
-        return kv.matchingColumnNoDelimiter(this.col);
+        return kv.matchingColumnNoDelimiter(this.col, this.familylength);
       } else if(this.matchType == MATCH_TYPE.FAMILY_ONLY) {
         return kv.matchingFamily(this.family);
       } else if (this.matchType == MATCH_TYPE.REGEX) {

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

@@ -309,7 +309,9 @@ public class Store implements HConstants {
         // Check this edit is for me. Also, guard against writing the speical
         // METACOLUMN info such as HBASE::CACHEFLUSH entries
         KeyValue kv = val.getKeyValue();
-        if (val.isTransactionEntry() || kv.matchingColumnNoDelimiter(HLog.METACOLUMN) ||
+        if (val.isTransactionEntry() ||
+            kv.matchingColumnNoDelimiter(HLog.METACOLUMN,
+              HLog.METACOLUMN.length - 1) ||
           !Bytes.equals(key.getRegionName(), regioninfo.getRegionName()) ||
           !kv.matchingFamily(family.getName())) {
           continue;

src/java/org/apache/hadoop/hbase/util/MetaUtils.java

@@ -251,7 +251,8 @@ public class MetaUtils {
         HRegionInfo info = null;
         for (KeyValue kv: results) {
           if (KeyValue.META_COMPARATOR.compareColumns(kv,
-            HConstants.COL_REGIONINFO, 0, HConstants.COL_REGIONINFO.length) == 0) {
+            HConstants.COL_REGIONINFO, 0, HConstants.COL_REGIONINFO.length,
+              HConstants.COLUMN_FAMILY_STR.length()) == 0) {
             info = Writables.getHRegionInfoOrNull(kv.getValue());
             if (info == null) {
               LOG.warn("region info is null for row " +

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

@@ -35,6 +35,19 @@ import org.apache.hadoop.hbase.util.Bytes;
 public class TestKeyValue extends TestCase {
   private final Log LOG = LogFactory.getLog(this.getClass().getName());
 
+  public void testColumnCompare() throws Exception {
+    final byte [] a = Bytes.toBytes("aaa");
+    byte [] column1 = Bytes.toBytes("abc:def");
+    byte [] column2 = Bytes.toBytes("abcd:ef");
+    KeyValue aaa = new KeyValue(a, column1, a);
+    assertFalse(KeyValue.COMPARATOR.
+      compareColumns(aaa, column2, 0, column2.length, 4) == 0);
+    column1 = Bytes.toBytes("abcd:");
+    aaa = new KeyValue(a, column1, a);
+    assertFalse(KeyValue.COMPARATOR.
+      compareColumns(aaa, column1, 0, column1.length, 4) == 0);
+  }
+
   public void testBasics() throws Exception {
     LOG.info("LOWKEY: " + KeyValue.LOWESTKEY.toString());
     check(Bytes.toBytes(getName()),