HBASE-1975 SingleColumnValueFilter: Add ability to match the value of previous versions of the specified column

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

CHANGES.txt

@@ -182,6 +182,9 @@ Release 0.21.0 - Unreleased
                a test and development environment (Wim Van Leuven via Stack)
    HBASE-1963  Output to multiple tables from Hadoop MR without use of HTable
                (Kevin Peterson via Andrew Purtell)
+   HBASE-1975  SingleColumnValueFilter: Add ability to match the value of
+               previous versions of the specified column
+               (Jeremiah Jacquet via Stack)
 
   OPTIMIZATIONS
    HBASE-410   [testing] Speed up the test suite

src/java/org/apache/hadoop/hbase/filter/SingleColumnValueFilter.java

@@ -48,10 +48,13 @@ import org.apache.hadoop.hbase.util.Bytes;
  * <p>
  * To prevent the entire row from being emitted if the column is not found
  * on a row, use {@link #setFilterIfMissing}.
- * <p>
  * Otherwise, if the column is found, the entire row will be emitted only if
  * the value passes.  If the value fails, the row will be filtered out.
  * <p>
+ * In order to test values of previous versions (timestamps), set
+ * {@link #setLatestVersionOnly} to false. The default is true, meaning that
+ * only the latest version's value is tested and all previous versions are ignored.
+ * <p>
  * To filter based on the value of all scanned columns, use {@link ValueFilter}.
  */
 public class SingleColumnValueFilter implements Filter {
@@ -64,7 +67,8 @@ public class SingleColumnValueFilter implements Filter {
   private boolean foundColumn = false;
   private boolean matchedColumn = false;
   private boolean filterIfMissing = false;
-  
+  private boolean latestVersionOnly = true;
+
   /**
    * Writable constructor, do not use.
    */
@@ -120,7 +124,7 @@ public class SingleColumnValueFilter implements Filter {
     if (this.matchedColumn) {
       // We already found and matched the single column, all keys now pass
       return ReturnCode.INCLUDE;
-    } else if (this.foundColumn) {
+    } else if (this.latestVersionOnly && this.foundColumn) {
       // We found but did not match the single column, skip to next row
       return ReturnCode.NEXT_ROW;
     }
@@ -130,7 +134,7 @@ public class SingleColumnValueFilter implements Filter {
     foundColumn = true;
     if (filterColumnValue(keyValue.getBuffer(),
         keyValue.getValueOffset(), keyValue.getValueLength())) {
-      return ReturnCode.NEXT_ROW;
+      return this.latestVersionOnly? ReturnCode.NEXT_ROW: ReturnCode.INCLUDE;
     }
     this.matchedColumn = true;
     return ReturnCode.INCLUDE;
@@ -196,6 +200,26 @@ public class SingleColumnValueFilter implements Filter {
     this.filterIfMissing = filterIfMissing;
   }
 
+  /**
+   * Get whether only the latest version of the column value should be compared.
+   * If true, the row will be returned if only the latest version of the column
+   * value matches. If false, the row will be returned if any version of the
+   * column value matches. The default is true.
+   */
+  public boolean getLatestVersionOnly() {
+    return latestVersionOnly;
+  }
+
+  /**
+   * Set whether only the latest version of the column value should be compared.
+   * If true, the row will be returned if only the latest version of the column
+   * value matches. If false, the row will be returned if any version of the
+   * column value matches. The default is true.
+   */
+  public void setLatestVersionOnly(boolean latestVersionOnly) {
+    this.latestVersionOnly = latestVersionOnly;
+  }
+
   public void readFields(final DataInput in) throws IOException {
     this.columnFamily = Bytes.readByteArray(in);
     if(this.columnFamily.length == 0) {
@@ -211,6 +235,7 @@ public class SingleColumnValueFilter implements Filter {
     this.foundColumn = in.readBoolean();
     this.matchedColumn = in.readBoolean();
     this.filterIfMissing = in.readBoolean();
+    this.latestVersionOnly = in.readBoolean();
   }
 
   public void write(final DataOutput out) throws IOException {
@@ -222,5 +247,6 @@ public class SingleColumnValueFilter implements Filter {
     out.writeBoolean(foundColumn);
     out.writeBoolean(matchedColumn);
     out.writeBoolean(filterIfMissing);
+    out.writeBoolean(latestVersionOnly);
   }
-}
+}

src/test/org/apache/hadoop/hbase/filter/TestSingleColumnValueFilter.java

@@ -77,7 +77,7 @@ public class TestSingleColumnValueFilter extends TestCase {
       new RegexStringComparator(QUICK_REGEX));
   }
 
-  private void basicFilterTests(Filter filter)
+  private void basicFilterTests(SingleColumnValueFilter filter)
       throws Exception {
     KeyValue kv = new KeyValue(ROW, COLUMN_FAMILY, COLUMN_QUALIFIER, VAL_2);
     assertTrue("basicFilter1", filter.filterKeyValue(kv) == Filter.ReturnCode.INCLUDE);
@@ -93,6 +93,13 @@ public class TestSingleColumnValueFilter extends TestCase {
     assertTrue("basicFilter4", filter.filterKeyValue(kv) == Filter.ReturnCode.NEXT_ROW);
     assertFalse("basicFilterAllRemaining", filter.filterAllRemaining());
     assertTrue("basicFilterNotNull", filter.filterRow());
+    filter.reset();
+    filter.setLatestVersionOnly(false);
+    kv = new KeyValue(ROW, COLUMN_FAMILY, COLUMN_QUALIFIER, VAL_1);
+    assertTrue("basicFilter5", filter.filterKeyValue(kv) == Filter.ReturnCode.INCLUDE);
+    kv = new KeyValue(ROW, COLUMN_FAMILY, COLUMN_QUALIFIER, VAL_2);
+    assertTrue("basicFilter5", filter.filterKeyValue(kv) == Filter.ReturnCode.INCLUDE);
+    assertFalse("basicFilterNotNull", filter.filterRow());
   }
 
   private void substrFilterTests(Filter filter) 
@@ -144,7 +151,7 @@ public class TestSingleColumnValueFilter extends TestCase {
    * @throws Exception
    */
   public void testStop() throws Exception {
-    basicFilterTests(basicFilter);
+    basicFilterTests((SingleColumnValueFilter)basicFilter);
     substrFilterTests(substrFilter);
     regexFilterTests(regexFilter);
   }                               
@@ -155,10 +162,10 @@ public class TestSingleColumnValueFilter extends TestCase {
    */                       
   public void testSerialization() throws Exception {
     Filter newFilter = serializationTest(basicFilter);
-    basicFilterTests(newFilter);
+    basicFilterTests((SingleColumnValueFilter)newFilter);
     newFilter = serializationTest(substrFilter);
     substrFilterTests(newFilter);
     newFilter = serializationTest(regexFilter);
     regexFilterTests(newFilter);
   }                   
-}
+}