HBASE-3191 FilterList with MUST_PASS_ONE and SCVF isn't working

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

CHANGES.txt

@@ -643,6 +643,8 @@ Release 0.21.0 - Unreleased
    HBASE-1932  Encourage use of 'lzo' compression... add the wiki page to
                getting started
    HBASE-3151  NPE when trying to read regioninfo from .META.
+   HBASE-3191  FilterList with MUST_PASS_ONE and SCVF isn't working
+               (Stefan Seelmann via Stack)
 
 
   IMPROVEMENTS

src/main/java/org/apache/hadoop/hbase/filter/FilterList.java

@@ -161,6 +161,8 @@ public class FilterList implements Filter {
 
   @Override
   public ReturnCode filterKeyValue(KeyValue v) {
+    ReturnCode rc = operator == Operator.MUST_PASS_ONE?
+        ReturnCode.SKIP: ReturnCode.INCLUDE;
     for (Filter filter : filters) {
       if (operator == Operator.MUST_PASS_ALL) {
         if (filter.filterAllRemaining()) {
@@ -180,15 +182,15 @@ public class FilterList implements Filter {
 
         switch (filter.filterKeyValue(v)) {
         case INCLUDE:
-          return ReturnCode.INCLUDE;
+          rc = ReturnCode.INCLUDE;
+          // must continue here to evaluate all filters
         case NEXT_ROW:
         case SKIP:
           // continue;
         }
       }
     }
-    return operator == Operator.MUST_PASS_ONE?
+    return rc;
-      ReturnCode.SKIP: ReturnCode.INCLUDE;
   }
 
   @Override

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

@@ -1041,6 +1041,82 @@ public class TestFilter extends HBaseTestCase {
     verifyScanFull(s, kvs);
   }
 
+  public void testFilterListWithSingleColumnValueFilter() throws IOException {
+    // Test for HBASE-3191
+
+    // Scan using SingleColumnValueFilter
+    SingleColumnValueFilter f1 = new SingleColumnValueFilter(FAMILIES[0], QUALIFIERS_ONE[0],
+          CompareOp.EQUAL, VALUES[0]);
+    f1.setFilterIfMissing( true );
+    Scan s1 = new Scan();
+    s1.addFamily(FAMILIES[0]);
+    s1.setFilter(f1);
+    KeyValue [] kvs1 = {
+        new KeyValue(ROWS_ONE[0], FAMILIES[0], QUALIFIERS_ONE[0], VALUES[0]),
+        new KeyValue(ROWS_ONE[0], FAMILIES[0], QUALIFIERS_ONE[2], VALUES[0]),
+        new KeyValue(ROWS_ONE[0], FAMILIES[0], QUALIFIERS_ONE[3], VALUES[0]),
+        new KeyValue(ROWS_ONE[2], FAMILIES[0], QUALIFIERS_ONE[0], VALUES[0]),
+        new KeyValue(ROWS_ONE[2], FAMILIES[0], QUALIFIERS_ONE[2], VALUES[0]),
+        new KeyValue(ROWS_ONE[2], FAMILIES[0], QUALIFIERS_ONE[3], VALUES[0]),
+        new KeyValue(ROWS_ONE[3], FAMILIES[0], QUALIFIERS_ONE[0], VALUES[0]),
+        new KeyValue(ROWS_ONE[3], FAMILIES[0], QUALIFIERS_ONE[2], VALUES[0]),
+        new KeyValue(ROWS_ONE[3], FAMILIES[0], QUALIFIERS_ONE[3], VALUES[0]),
+    };
+    verifyScanNoEarlyOut(s1, 3, 3);
+    verifyScanFull(s1, kvs1);
+
+    // Scan using another SingleColumnValueFilter, expect disjoint result
+    SingleColumnValueFilter f2 = new SingleColumnValueFilter(FAMILIES[0], QUALIFIERS_TWO[0],
+        CompareOp.EQUAL, VALUES[1]);
+    f2.setFilterIfMissing( true );
+    Scan s2 = new Scan();
+    s2.addFamily(FAMILIES[0]);
+    s2.setFilter(f2);
+    KeyValue [] kvs2 = {
+        new KeyValue(ROWS_TWO[0], FAMILIES[0], QUALIFIERS_TWO[0], VALUES[1]),
+        new KeyValue(ROWS_TWO[0], FAMILIES[0], QUALIFIERS_TWO[2], VALUES[1]),
+        new KeyValue(ROWS_TWO[0], FAMILIES[0], QUALIFIERS_TWO[3], VALUES[1]),
+        new KeyValue(ROWS_TWO[2], FAMILIES[0], QUALIFIERS_TWO[0], VALUES[1]),
+        new KeyValue(ROWS_TWO[2], FAMILIES[0], QUALIFIERS_TWO[2], VALUES[1]),
+        new KeyValue(ROWS_TWO[2], FAMILIES[0], QUALIFIERS_TWO[3], VALUES[1]),
+        new KeyValue(ROWS_TWO[3], FAMILIES[0], QUALIFIERS_TWO[0], VALUES[1]),
+        new KeyValue(ROWS_TWO[3], FAMILIES[0], QUALIFIERS_TWO[2], VALUES[1]),
+        new KeyValue(ROWS_TWO[3], FAMILIES[0], QUALIFIERS_TWO[3], VALUES[1]),
+    };
+    verifyScanNoEarlyOut(s2, 3, 3);
+    verifyScanFull(s2, kvs2);
+
+    // Scan, ORing the two previous filters, expect unified result
+    FilterList f = new FilterList(Operator.MUST_PASS_ONE);
+    f.addFilter(f1);
+    f.addFilter(f2);
+    Scan s = new Scan();
+    s.addFamily(FAMILIES[0]);
+    s.setFilter(f);
+    KeyValue [] kvs = {
+        new KeyValue(ROWS_ONE[0], FAMILIES[0], QUALIFIERS_ONE[0], VALUES[0]),
+        new KeyValue(ROWS_ONE[0], FAMILIES[0], QUALIFIERS_ONE[2], VALUES[0]),
+        new KeyValue(ROWS_ONE[0], FAMILIES[0], QUALIFIERS_ONE[3], VALUES[0]),
+        new KeyValue(ROWS_ONE[2], FAMILIES[0], QUALIFIERS_ONE[0], VALUES[0]),
+        new KeyValue(ROWS_ONE[2], FAMILIES[0], QUALIFIERS_ONE[2], VALUES[0]),
+        new KeyValue(ROWS_ONE[2], FAMILIES[0], QUALIFIERS_ONE[3], VALUES[0]),
+        new KeyValue(ROWS_ONE[3], FAMILIES[0], QUALIFIERS_ONE[0], VALUES[0]),
+        new KeyValue(ROWS_ONE[3], FAMILIES[0], QUALIFIERS_ONE[2], VALUES[0]),
+        new KeyValue(ROWS_ONE[3], FAMILIES[0], QUALIFIERS_ONE[3], VALUES[0]),
+        new KeyValue(ROWS_TWO[0], FAMILIES[0], QUALIFIERS_TWO[0], VALUES[1]),
+        new KeyValue(ROWS_TWO[0], FAMILIES[0], QUALIFIERS_TWO[2], VALUES[1]),
+        new KeyValue(ROWS_TWO[0], FAMILIES[0], QUALIFIERS_TWO[3], VALUES[1]),
+        new KeyValue(ROWS_TWO[2], FAMILIES[0], QUALIFIERS_TWO[0], VALUES[1]),
+        new KeyValue(ROWS_TWO[2], FAMILIES[0], QUALIFIERS_TWO[2], VALUES[1]),
+        new KeyValue(ROWS_TWO[2], FAMILIES[0], QUALIFIERS_TWO[3], VALUES[1]),
+        new KeyValue(ROWS_TWO[3], FAMILIES[0], QUALIFIERS_TWO[0], VALUES[1]),
+        new KeyValue(ROWS_TWO[3], FAMILIES[0], QUALIFIERS_TWO[2], VALUES[1]),
+        new KeyValue(ROWS_TWO[3], FAMILIES[0], QUALIFIERS_TWO[3], VALUES[1]),
+    };
+    verifyScanNoEarlyOut(s, 6, 3);
+    verifyScanFull(s, kvs);
+  }
+
   public void testSingleColumnValueFilter() throws IOException {
 
     // From HBASE-1821