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

fix range query buglet; add debug logging 

patch by jbellis; tested by Mark Robson for CASSANDRA-348

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@801676 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-08-06 15:34:22 +00:00
parent e1e6a6c554
commit 449a444c7d
4 changed files with 69 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/java/org/apache/cassandra/db/RangeReply.java
View File

@ -21,6 +21,8 @@ package org.apache.cassandra.db;
import java.util.*;
import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.DataOutputBuffer;
import org.apache.cassandra.net.Message;
@ -50,6 +52,15 @@ public class RangeReply
return originalMessage.getReply(StorageService.getLocalStorageEndPoint(), data);
}
@Override
public String toString()
{
return "RangeReply(" +
"keys=[" + StringUtils.join(keys, ", ") +
"], completed=" + rangeCompletedLocally +
')';
}
public static RangeReply read(byte[] body) throws IOException
{
DataInputBuffer bufIn = new DataInputBuffer();

9
src/java/org/apache/cassandra/service/CassandraServer.java
View File

@ -550,7 +550,14 @@ public class CassandraServer implements Cassandra.Iface
throw new InvalidRequestException("maxResults must be positive");
}
return StorageProxy.getKeyRange(new RangeCommand(tablename, columnFamily, startWith, stopAt, maxResults));
try
{
return StorageProxy.getKeyRange(new RangeCommand(tablename, columnFamily, startWith, stopAt, maxResults));
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
// main method moved to CassandraDaemon

9
src/java/org/apache/cassandra/service/RangeVerbHandler.java
View File

@ -18,6 +18,8 @@
*/
package org.apache.cassandra.service;
import org.apache.log4j.Logger;
import org.apache.cassandra.db.RangeCommand;
import org.apache.cassandra.db.RangeReply;
import org.apache.cassandra.db.Table;
@ -27,16 +29,19 @@ import org.apache.cassandra.net.MessagingService;
public class RangeVerbHandler implements IVerbHandler
{
private static final Logger logger = Logger.getLogger(RangeVerbHandler.class);
public void doVerb(Message message)
{
RangeReply rangeReply;
try
{
RangeCommand command = RangeCommand.read(message);
Table table = Table.open(command.table);
rangeReply = table.getKeyRange(command.columnFamily, command.startWith, command.stopAt, command.maxResults);
RangeReply rangeReply = table.getKeyRange(command.columnFamily, command.startWith, command.stopAt, command.maxResults);
Message response = rangeReply.getReply(message);
if (logger.isDebugEnabled())
logger.debug("Sending " + rangeReply + " to " + message.getMessageId() + "@" + message.getFrom());
MessagingService.getMessagingInstance().sendOneWay(response, message.getFrom());
}
catch (Exception e)

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

@ -359,14 +359,6 @@ public class StorageProxy implements StorageProxyMBean
/**
* This is a multiget version of the above method.
* @param tablename
* @param keys
* @param columnFamily
* @param start
* @param count
* @return
* @throws IOException
* @throws TimeoutException
*/
public static Map<String, Row> strongReadProtocol(String[] keys, ReadCommand readCommand) throws IOException, TimeoutException
{
@ -571,10 +563,6 @@ public class StorageProxy implements StorageProxyMBean
* This version is used when results for multiple keys needs to be
* retrieved.
*
* @param tablename name of the table that needs to be queried
* @param keys keys whose values we are interested in
* @param columnFamily name of the "column" we are interested in
* @param columns the columns we are interested in
* @return a mapping of key --> Row
* @throws Exception
*/
@ -631,64 +619,61 @@ public class StorageProxy implements StorageProxyMBean
return row;
}
static List<String> getKeyRange(RangeCommand command)
static List<String> getKeyRange(RangeCommand command) throws IOException
{
long startTime = System.currentTimeMillis();
int endpointOffset = 0;
List<String> allKeys = new ArrayList<String>();
int maxResults = command.maxResults;
try
{
EndPoint endPoint = StorageService.instance().findSuitableEndPoint(command.startWith, endpointOffset);
String firstEndpoint = endPoint.toString();
EndPoint endPoint = StorageService.instance().findSuitableEndPoint(command.startWith, endpointOffset);
String firstEndpoint = endPoint.toString();
do
do
{
Message message = command.getMessage();
if (logger.isDebugEnabled())
logger.debug("reading " + command + " from " + message.getMessageId() + "@" + endPoint);
IAsyncResult iar = MessagingService.getMessagingInstance().sendRR(message, endPoint);
// read response
byte[] responseBody = new byte[0];
try
{
IAsyncResult iar = MessagingService.getMessagingInstance().sendRR(command.getMessage(), endPoint);
responseBody = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
RangeReply rangeReply = RangeReply.read(responseBody);
List<String> rangeKeys = rangeReply.keys;
// read response
byte[] responseBody = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
RangeReply rangeReply = RangeReply.read(responseBody);
List<String> rangeKeys = rangeReply.keys;
// deal with key overlaps
if (allKeys.size() > 0 && rangeKeys != null && rangeKeys.size() > 0 && allKeys.get(allKeys.size() - 1).equals(rangeKeys.get(0)))
{
allKeys.remove(allKeys.size() - 1);
allKeys.addAll(rangeKeys);
}
else if (rangeKeys != null && rangeKeys.size() > 0)
{
allKeys.addAll(rangeKeys);
}
// deal with key overlaps
if (allKeys.size() > 0 && rangeKeys != null && rangeKeys.size() > 0 && allKeys.get(allKeys.size() - 1).equals(rangeKeys.get(0)))
{
allKeys.remove(allKeys.size() - 1);
allKeys.addAll(rangeKeys);
}
else if (rangeKeys != null && rangeKeys.size() > 0)
{
allKeys.addAll(rangeKeys);
}
if (allKeys.size() >= maxResults || rangeReply.rangeCompletedLocally)
{
break;
}
if (allKeys.size() >= maxResults || rangeReply.rangeCompletedLocally)
{
break;
}
String newStartWith = (allKeys.size() > 0) ? allKeys.get(allKeys.size() - 1) : command.startWith;
command = new RangeCommand(command.table, command.columnFamily, newStartWith, command.stopAt, command.maxResults - allKeys.size());
endPoint = StorageService.instance().findSuitableEndPoint(command.startWith, ++endpointOffset);
} while (!endPoint.toString().equals(firstEndpoint));
String newStartAt = (allKeys.size() > 0) ? allKeys.get(allKeys.size() - 1) : command.stopAt;
command = new RangeCommand(command.table, command.columnFamily,
newStartAt, command.stopAt,
command.maxResults - rangeKeys.size());
endPoint = StorageService.instance().findSuitableEndPoint(command.startWith, ++endpointOffset);
} while (!endPoint.toString().equals(firstEndpoint));
return (allKeys.size() > maxResults)
? allKeys.subList(0, maxResults)
: allKeys;
}
catch (Exception e)
{
throw new RuntimeException("error reading keyrange " + command, e);
}
finally
{
rangeStats.add(System.currentTimeMillis() - startTime);
}
rangeStats.add(System.currentTimeMillis() - startTime);
return (allKeys.size() > maxResults)
? allKeys.subList(0, maxResults)
: allKeys;
}
public double getReadLatency()