diff --git a/CHANGES.txt b/CHANGES.txt index 4a8ee42037..1e51f5489b 100644 --- a/CHANGES.txt +++ b/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) diff --git a/NEWS.txt b/NEWS.txt index e685b126c2..0b227f3170 100644 --- a/NEWS.txt +++ b/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 diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 98223f7d0e..7ff46c9b89 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1304,6 +1304,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public List filter(AbstractScanIterator rowIterator, ExtendedFilter filter) { + if (logger.isDebugEnabled()) + logger.debug("Filtering {} for rows matching {}", rowIterator, filter); List rows = new ArrayList(); int columnsCount = 0; try diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java index c13e88123d..8abc46019f 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java @@ -124,7 +124,7 @@ public class ColumnFamilyInputFormat extends InputFormat 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"; diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java index cbe2b3c090..46b767a6aa 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java @@ -68,6 +68,7 @@ public class ColumnFamilyRecordReader extends RecordReader filter; public ColumnFamilyRecordReader() { @@ -131,6 +132,8 @@ public class ColumnFamilyRecordReader extends RecordReader 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 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) { diff --git a/src/java/org/apache/cassandra/thrift/ThriftValidation.java b/src/java/org/apache/cassandra/thrift/ThriftValidation.java index 31035e3836..b1ab665655 100644 --- a/src/java/org/apache/cassandra/thrift/ThriftValidation.java +++ b/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) {