natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

add option to skip start key in range query (StorageProxy only for now) and test. 

patch by jbellis; reviewed by stuhood for CASSANDRA-759

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@906272 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-02-03 22:44:56 +00:00
parent 30f0216a8f
commit ecdcc00ff3
6 changed files with 67 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -39,7 +39,6 @@ import org.apache.cassandra.cache.JMXInstrumentedCache;
import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.*;
import org.apache.cassandra.io.util.FileUtils;
@ -931,13 +930,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* @param startWith key to start with, inclusive. empty string = start at beginning.
* @param stopAt key to stop at, inclusive. empty string = stop only when keys are exhausted.
* @param maxResults
* @param includeStartKey
* @return list of keys between startWith and stopAt
TODO refactor better. this is just getKeyRange w/o the deletion check, for the benefit of
range_slice. still opens one randomaccessfile per key, which sucks. something like compactioniterator
would be better.
*/
private boolean getKeyRange(List<String> keys, final DecoratedKey startWith, final DecoratedKey stopAt, int maxResults)
private boolean getKeyRange(List<String> keys, final DecoratedKey startWith, final DecoratedKey stopAt, int maxResults, boolean includeStartKey)
throws IOException, ExecutionException, InterruptedException
{
// getKeyRange requires start <= stop. getRangeSlice handles range wrapping if necessary.
@ -1011,14 +1011,20 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
try
{
// pull keys out of the CollatedIterator
boolean rangeCompletedLocally = false;
boolean first = true;
for (DecoratedKey current : reduced)
{
if (!stopAt.isEmpty() && stopAt.compareTo(current) < 0)
{
return true;
}
keys.add(current.key);
if (includeStartKey || !first || !current.equals(startWith))
{
keys.add(current.key);
}
first = false;
if (keys.size() >= maxResults)
{
return true;
@ -1046,27 +1052,28 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* @param keyMax maximum number of keys to process, regardless of startKey/finishKey
* @param sliceRange may be null if columnNames is specified. specifies contiguous columns to return in what order.
* @param columnNames may be null if sliceRange is specified. specifies which columns to return in what order. @return list of key->list<column> tuples.
* @param includeStartKey
* @throws IOException
* @throws ExecutionException
* @throws InterruptedException
*/
public RangeSliceReply getRangeSlice(byte[] super_column, final DecoratedKey startKey, final DecoratedKey finishKey, int keyMax, SliceRange sliceRange, List<byte[]> columnNames)
public RangeSliceReply getRangeSlice(byte[] super_column, final DecoratedKey startKey, final DecoratedKey finishKey, int keyMax, SliceRange sliceRange, List<byte[]> columnNames, boolean includeStartKey)
throws IOException, ExecutionException, InterruptedException
{
List<String> keys = new ArrayList<String>();
boolean completed;
if (finishKey.isEmpty() || startKey.compareTo(finishKey) <= 0)
{
completed = getKeyRange(keys, startKey, finishKey, keyMax);
completed = getKeyRange(keys, startKey, finishKey, keyMax, includeStartKey);
}
else
{
// wrapped range
DecoratedKey emptyKey = new DecoratedKey(StorageService.getPartitioner().getMinimumToken(), null);
completed = getKeyRange(keys, startKey, emptyKey, keyMax);
completed = getKeyRange(keys, startKey, emptyKey, keyMax, includeStartKey);
if (!completed)
{
completed = getKeyRange(keys, emptyKey, finishKey, keyMax);
completed = getKeyRange(keys, emptyKey, finishKey, keyMax, true);
}
}
List<Row> rows = new ArrayList<Row>(keys.size());

24
src/java/org/apache/cassandra/db/RangeSliceCommand.java
View File

@ -38,8 +38,6 @@ package org.apache.cassandra.db;
import org.apache.cassandra.concurrent.StageManager;
import static org.apache.cassandra.thrift.ThriftGlue.createColumnParent;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.net.Message;
@ -71,19 +69,14 @@ public class RangeSliceCommand
public final DecoratedKey startKey;
public final DecoratedKey finishKey;
public final int max_keys;
public final boolean includeStartKey;
public RangeSliceCommand(String keyspace, ColumnParent column_parent, SlicePredicate predicate, DecoratedKey startKey, DecoratedKey finishKey, int max_keys)
{
this.keyspace = keyspace;
column_family = column_parent.getColumn_family();
super_column = column_parent.getSuper_column();
this.predicate = predicate;
this.startKey = startKey;
this.finishKey = finishKey;
this.max_keys = max_keys;
this(keyspace, column_parent.getColumn_family(), column_parent.getSuper_column(), predicate, startKey, finishKey, max_keys, true);
}
public RangeSliceCommand(String keyspace, String column_family, byte[] super_column, SlicePredicate predicate, DecoratedKey startKey, DecoratedKey finishKey, int max_keys)
public RangeSliceCommand(String keyspace, String column_family, byte[] super_column, SlicePredicate predicate, DecoratedKey startKey, DecoratedKey finishKey, int max_keys, boolean includeStartKey)
{
this.keyspace = keyspace;
this.column_family = column_family;
@ -92,6 +85,7 @@ public class RangeSliceCommand
this.startKey = startKey;
this.finishKey = finishKey;
this.max_keys = max_keys;
this.includeStartKey = includeStartKey;
}
public Message getMessage() throws IOException
@ -127,6 +121,7 @@ class SliceCommandSerializer implements ICompactSerializer<RangeSliceCommand>
DecoratedKey.serializer().serialize(sliceCommand.startKey, dos);
DecoratedKey.serializer().serialize(sliceCommand.finishKey, dos);
dos.writeInt(sliceCommand.max_keys);
dos.writeBoolean(sliceCommand.includeStartKey);
}
public RangeSliceCommand deserialize(DataInputStream dis) throws IOException
@ -146,13 +141,8 @@ class SliceCommandSerializer implements ICompactSerializer<RangeSliceCommand>
DecoratedKey startKey = DecoratedKey.serializer().deserialize(dis);
DecoratedKey finishKey = DecoratedKey.serializer().deserialize(dis);
int max_keys = dis.readInt();
return new RangeSliceCommand(keyspace,
createColumnParent(column_family, super_column),
pred,
startKey,
finishKey,
max_keys);
boolean includeStartKey = dis.readBoolean();
return new RangeSliceCommand(keyspace, column_family, super_column, pred, startKey, finishKey, max_keys, includeStartKey);
}
static byte[] readBuf(int len, DataInputStream dis) throws IOException

16
src/java/org/apache/cassandra/service/RangeSliceVerbHandler.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.service;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.RangeSliceCommand;
import org.apache.cassandra.db.RangeSliceReply;
import org.apache.cassandra.db.Table;
@ -36,13 +37,14 @@ public class RangeSliceVerbHandler implements IVerbHandler
try
{
RangeSliceCommand command = RangeSliceCommand.read(message);
RangeSliceReply reply = Table.open(command.keyspace).getColumnFamilyStore(command.column_family).getRangeSlice(
command.super_column,
command.startKey,
command.finishKey,
command.max_keys,
command.predicate.slice_range,
command.predicate.column_names);
ColumnFamilyStore cfs = Table.open(command.keyspace).getColumnFamilyStore(command.column_family);
RangeSliceReply reply = cfs.getRangeSlice(command.super_column,
command.startKey,
command.finishKey,
command.max_keys,
command.predicate.slice_range,
command.predicate.column_names,
command.includeStartKey);
Message response = reply.getReply(message);
if (logger.isDebugEnabled())
logger.debug("Sending " + reply+ " to " + message.getMessageId() + "@" + message.getFrom());

2
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -564,7 +564,7 @@ public class StorageProxy implements StorageProxyMBean
? new DecoratedKey<Token<?>>(primaryRange.right, null)
: (DecoratedKey<?>) ObjectUtils.min(command.finishKey, new DecoratedKey<Token<?>>(primaryRange.right, null));
}
RangeSliceCommand c2 = new RangeSliceCommand(command.keyspace, command.column_family, command.super_column, command.predicate, startKey, finishKey, command.max_keys);
RangeSliceCommand c2 = new RangeSliceCommand(command.keyspace, command.column_family, command.super_column, command.predicate, startKey, finishKey, command.max_keys, command.includeStartKey);
Message message = c2.getMessage();
// collect replies and resolve according to consistency level

3
test/unit/org/apache/cassandra/Util.java
View File

@ -69,7 +69,8 @@ public class Util
emptyKey,
10000,
new SliceRange(ArrayUtils.EMPTY_BYTE_ARRAY, ArrayUtils.EMPTY_BYTE_ARRAY, false, 10000),
null);
null,
true);
}
/**

38
test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
View File

@ -30,7 +30,6 @@ import static junit.framework.Assert.assertEquals;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.Util;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.utils.WrappedRunnable;
import java.net.InetAddress;
@ -133,6 +132,38 @@ public class ColumnFamilyStoreTest extends CleanupHelper
@Test
public void testWrappedRangeQuery() throws IOException, ExecutionException, InterruptedException
{
ColumnFamilyStore cfs = insertKey1Key2();
IPartitioner p = StorageService.getPartitioner();
RangeSliceReply result = cfs.getRangeSlice(ArrayUtils.EMPTY_BYTE_ARRAY,
p.decorateKey("key2"),
p.decorateKey("key1"),
10,
null,
Arrays.asList("asdf".getBytes()),
true);
assertEquals(2, result.rows.size());
}
@Test
public void testSkipStartKey() throws IOException, ExecutionException, InterruptedException
{
ColumnFamilyStore cfs = insertKey1Key2();
IPartitioner p = StorageService.getPartitioner();
RangeSliceReply result = cfs.getRangeSlice(ArrayUtils.EMPTY_BYTE_ARRAY,
p.decorateKey("key1"),
p.decorateKey("key2"),
10,
null,
Arrays.asList("asdf".getBytes()),
false);
assertEquals(1, result.rows.size());
assert result.rows.get(0).key.equals("key2");
}
private ColumnFamilyStore insertKey1Key2() throws IOException, ExecutionException, InterruptedException
{
List<RowMutation> rms = new LinkedList<RowMutation>();
RowMutation rm;
@ -145,9 +176,6 @@ public class ColumnFamilyStoreTest extends CleanupHelper
rm.add(new QueryPath("Standard1", null, "Column1".getBytes()), "asdf".getBytes(), 0);
rms.add(rm);
ColumnFamilyStore cfs = Util.writeColumnFamily(rms);
IPartitioner p = StorageService.getPartitioner();
RangeSliceReply result = cfs.getRangeSlice(ArrayUtils.EMPTY_BYTE_ARRAY, p.decorateKey("key2"), p.decorateKey("key1"), 10, null, Arrays.asList("asdf".getBytes()));
assertEquals(2, result.rows.size());
return cfs;
}
}