propagate range filter to ColumnFamilyRecordReader

patch by jbellis; reviewed by brandonwilliams for CASSANDRA-2878

CHANGES.txt

@@ -20,6 +20,7 @@
  * Use faster bytes comparison (CASSANDRA-3434)
  * Bulk loader is no longer a fat client, (HADOOP) bulk load output format
    (CASSANDRA-3045)
+ * (Hadoop) add support for KeyRange.filter
  * remove assumption that keys and token are in bijection
    (CASSANDRA-1034, 3574, 3604)
  * always remove endpoints from delevery queue in HH (CASSANDRA-3546)

NEWS.txt

@@ -55,6 +55,7 @@ Features
       be pinned to specfic media.
     - Hadoop: a new BulkOutputFormat is included which will directly write
       SSTables locally and then stream them into the cluster.
+    - Hadoop: KeyRange.filter is now supported with ColumnFamilyInputFormat
     - The bulk loader is not longer a fat client; it can be run from an
       existing machine in a cluster.
     - A new write survey mode has been added, similar to bootstrap (enabled via

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -1304,6 +1304,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public List<Row> filter(AbstractScanIterator rowIterator, ExtendedFilter filter)
     {
+        if (logger.isDebugEnabled())
+            logger.debug("Filtering {} for rows matching {}", rowIterator, filter);
          List<Row> rows = new ArrayList<Row>();
          int columnsCount = 0;
          try

src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java

@@ -124,7 +124,7 @@ public class ColumnFamilyInputFormat extends InputFormat<ByteBuffer, SortedMap<B
             KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf);
             IPartitioner partitioner = null;
             Range<Token> jobRange = null;
-            if (jobKeyRange != null)
+            if (jobKeyRange != null && jobKeyRange.start_token != null)
             {
                 partitioner = ConfigHelper.getInputPartitioner(context.getConfiguration());
                 assert partitioner.preservesOrder() : "ConfigHelper.setInputKeyRange(..) can only be used with a order preserving paritioner";

src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java

@@ -68,6 +68,7 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
     private Cassandra.Client client;
     private ConsistencyLevel consistencyLevel;
     private int keyBufferSize = 8192;
+    private List<IndexExpression> filter;
 
     public ColumnFamilyRecordReader()
     {
@@ -131,6 +132,8 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
     {
         this.split = (ColumnFamilySplit) split;
         Configuration conf = context.getConfiguration();
+        KeyRange jobRange = ConfigHelper.getInputKeyRange(conf);
+        filter = jobRange == null ? null : jobRange.row_filter;
         predicate = ConfigHelper.getInputSlicePredicate(conf);
         isEmptyPredicate = isEmptyPredicate(predicate);
         totalRowCount = ConfigHelper.getInputSplitSize(conf);
@@ -283,7 +286,8 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
             
             KeyRange keyRange = new KeyRange(batchRowCount)
                                 .setStart_token(startToken)
-                                .setEnd_token(split.getEndToken());
+                                .setEnd_token(split.getEndToken())
+                                .setRow_filter(filter);
             try
             {
                 rows = client.get_range_slices(new ColumnParent(cfName),

src/java/org/apache/cassandra/hadoop/ConfigHelper.java

@@ -25,10 +25,7 @@ import java.util.List;
 
 import org.apache.cassandra.config.ConfigurationException;
 import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.thrift.Cassandra;
-import org.apache.cassandra.thrift.KeyRange;
-import org.apache.cassandra.thrift.SlicePredicate;
-import org.apache.cassandra.thrift.TBinaryProtocol;
+import org.apache.cassandra.thrift.*;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Hex;
 import org.apache.hadoop.conf.Configuration;
@@ -225,6 +222,26 @@ public class ConfigHelper
         conf.set(INPUT_KEYRANGE_CONFIG, thriftToString(range));
     }
 
+    /**
+     * Set the KeyRange to limit the rows.
+     * @param conf Job configuration you are about to run
+     */
+    public static void setInputRange(Configuration conf, String startToken, String endToken, List<IndexExpression> filter)
+    {
+        KeyRange range = new KeyRange().setStart_token(startToken).setEnd_token(endToken).setRow_filter(filter);
+        conf.set(INPUT_KEYRANGE_CONFIG, thriftToString(range));
+    }
+
+    /**
+     * Set the KeyRange to limit the rows.
+     * @param conf Job configuration you are about to run
+     */
+    public static void setInputRange(Configuration conf, List<IndexExpression> filter)
+    {
+        KeyRange range = new KeyRange().setRow_filter(filter);
+        conf.set(INPUT_KEYRANGE_CONFIG, thriftToString(range));
+    }
+
     /** may be null if unset */
     public static KeyRange getInputKeyRange(Configuration conf)
     {

src/java/org/apache/cassandra/thrift/ThriftValidation.java

@@ -514,13 +514,6 @@ public class ThriftValidation
         {
             throw new InvalidRequestException("super columns are not yet supported for indexing");
         }
-        if (!isEmpty(range.row_filter) && range.start_key == null)
-        {
-            // TODO: our current KEYS indexes can't do that efficiently
-            // (without scanning *all* the keys in the range and simply applying the filter to discard them when they don't match)
-            // See KeySearcher.search()
-            throw new InvalidRequestException("filtered queries must use concrete keys rather than tokens");
-        }
 
         if (range.count <= 0)
         {