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

clean up ReadCommand; it's basically a struct so use public final fields 

patch by jbellis; reviewed by Eric Evans for #88

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@766839 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-04-20 20:05:29 +00:00
parent 8e72ac4f9f
commit df95fb88dc
6 changed files with 160 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

264
src/org/apache/cassandra/db/ReadCommand.java
View File

@ -22,9 +22,10 @@ import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Arrays;
import java.util.Collections;
import org.apache.commons.lang.StringUtils;
@ -37,197 +38,162 @@ import org.apache.cassandra.service.StorageService;
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
*/
public class ReadCommand implements Serializable
public class ReadCommand
{
private static ICompactSerializer<ReadCommand> serializer_;
public static final String doRepair_ = "READ-REPAIR";
public static final String DO_REPAIR = "READ-REPAIR";
static
private static ReadCommandSerializer serializer = new ReadCommandSerializer();
public static ReadCommandSerializer serializer()
{
serializer_ = new ReadCommandSerializer();
return serializer;
}
static ICompactSerializer<ReadCommand> serializer()
private static List<String> EMPTY_COLUMNS = Arrays.asList(new String[0]);
public Message makeReadMessage() throws IOException
{
return serializer_;
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(bos);
ReadCommand.serializer().serialize(this, dos);
return new Message(StorageService.getLocalStorageEndPoint(), StorageService.readStage_, StorageService.readVerbHandler_, bos.toByteArray());
}
public static Message makeReadMessage(ReadCommand readCommand) throws IOException
public final String table;
public final String key;
public final String columnFamilyColumn;
public final int start;
public final int count;
public final long sinceTimestamp;
public final List<String> columnNames;
private boolean isDigestQuery = false;
public ReadCommand(String table, String key, String columnFamilyColumn, int start, int count, long sinceTimestamp, List<String> columnNames)
{
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream( bos );
ReadCommand.serializer().serialize(readCommand, dos);
Message message = new Message(StorageService.getLocalStorageEndPoint(), StorageService.readStage_, StorageService.readVerbHandler_, new Object[]{bos.toByteArray()});
return message;
this.table = table;
this.key = key;
this.columnFamilyColumn = columnFamilyColumn;
this.start = start;
this.count = count;
this.sinceTimestamp = sinceTimestamp;
this.columnNames = Collections.unmodifiableList(columnNames);
}
private String table_;
private String key_;
private String columnFamily_column_ = null;
private int start_ = -1;
private int count_ = -1 ;
private long sinceTimestamp_ = -1 ;
private List<String> columns_ = new ArrayList<String>();
private boolean isDigestQuery_ = false;
private ReadCommand()
{
}
public ReadCommand(String table, String key)
{
table_ = table;
key_ = key;
this(table, key, null, -1, -1, -1, EMPTY_COLUMNS);
}
public ReadCommand(String table, String key, String columnFamily_column)
public ReadCommand(String table, String key, String columnFamilyColumn)
{
table_ = table;
key_ = key;
columnFamily_column_ = columnFamily_column;
}
public ReadCommand(String table, String key, String columnFamily, List<String> columns)
{
table_ = table;
key_ = key;
columnFamily_column_ = columnFamily;
columns_ = columns;
}
public ReadCommand(String table, String key, String columnFamily_column, int start, int count)
{
table_ = table;
key_ = key;
columnFamily_column_ = columnFamily_column;
start_ = start ;
count_ = count;
this(table, key, columnFamilyColumn, -1, -1, -1, EMPTY_COLUMNS);
}
public ReadCommand(String table, String key, String columnFamily_column, long sinceTimestamp)
public ReadCommand(String table, String key, String columnFamilyColumn, List<String> columnNames)
{
table_ = table;
key_ = key;
columnFamily_column_ = columnFamily_column;
sinceTimestamp_ = sinceTimestamp ;
this(table, key, columnFamilyColumn, -1, -1, -1, columnNames);
}
String table()
public ReadCommand(String table, String key, String columnFamilyColumn, int start, int count)
{
return table_;
}
public String key()
{
return key_;
this(table, key, columnFamilyColumn, start, count, -1, EMPTY_COLUMNS);
}
String columnFamily_column()
public ReadCommand(String table, String key, String columnFamilyColumn, long sinceTimestamp)
{
return columnFamily_column_;
}
int start()
{
return start_;
}
int count()
{
return count_;
}
long sinceTimestamp()
{
return sinceTimestamp_;
this(table, key, columnFamilyColumn, -1, -1, sinceTimestamp, EMPTY_COLUMNS);
}
public boolean isDigestQuery()
{
return isDigestQuery_;
return isDigestQuery;
}
public void setIsDigestQuery(boolean isDigestQuery)
public void setDigestQuery(boolean isDigestQuery)
{
isDigestQuery_ = isDigestQuery;
this.isDigestQuery = isDigestQuery;
}
public List<String> getColumnNames()
public ReadCommand copy()
{
return columns_;
return new ReadCommand(table, key, columnFamilyColumn, start, count, sinceTimestamp, columnNames);
}
public String toString()
{
return "ReadMessage(" +
"table='" + table_ + '\'' +
", key='" + key_ + '\'' +
", columnFamily_column='" + columnFamily_column_ + '\'' +
", start=" + start_ +
", count=" + count_ +
", sinceTimestamp=" + sinceTimestamp_ +
", columns=[" + StringUtils.join(columns_, ", ") + "]" +
", isDigestQuery=" + isDigestQuery_ +
"table='" + table + '\'' +
", key='" + key + '\'' +
", columnFamilyColumn='" + columnFamilyColumn + '\'' +
", start=" + start +
", count=" + count +
", sinceTimestamp=" + sinceTimestamp +
", columns=[" + StringUtils.join(columnNames, ", ") + "]" +
')';
}
}
class ReadCommandSerializer implements ICompactSerializer<ReadCommand>
{
public void serialize(ReadCommand rm, DataOutputStream dos) throws IOException
{
dos.writeUTF(rm.table());
dos.writeUTF(rm.key());
dos.writeUTF(rm.columnFamily_column());
dos.writeInt(rm.start());
dos.writeInt(rm.count());
dos.writeLong(rm.sinceTimestamp());
dos.writeBoolean(rm.isDigestQuery());
List<String> columns = rm.getColumnNames();
dos.writeInt(columns.size());
if ( columns.size() > 0 )
{
for ( String column : columns )
{
dos.writeInt(column.getBytes().length);
dos.write(column.getBytes());
}
}
}
public void serialize(ReadCommand rm, DataOutputStream dos) throws IOException
{
dos.writeUTF(rm.table);
dos.writeUTF(rm.key);
dos.writeUTF(rm.columnFamilyColumn);
dos.writeInt(rm.start);
dos.writeInt(rm.count);
dos.writeLong(rm.sinceTimestamp);
dos.writeBoolean(rm.isDigestQuery());
dos.writeInt(rm.columnNames.size());
if (rm.columnNames.size() > 0)
{
for (String cName : rm.columnNames)
{
dos.writeInt(cName.getBytes().length);
dos.write(cName.getBytes());
}
}
}
public ReadCommand deserialize(DataInputStream dis) throws IOException
{
String table = dis.readUTF();
String key = dis.readUTF();
String columnFamily_column = dis.readUTF();
int start = dis.readInt();
int count = dis.readInt();
long sinceTimestamp = dis.readLong();
boolean isDigest = dis.readBoolean();
int size = dis.readInt();
List<String> columns = new ArrayList<String>();
for ( int i = 0; i < size; ++i )
{
byte[] bytes = new byte[dis.readInt()];
dis.readFully(bytes);
columns.add( new String(bytes) );
}
ReadCommand rm = null;
if ( columns.size() > 0 )
{
rm = new ReadCommand(table, key, columnFamily_column, columns);
}
else if( sinceTimestamp > 0 )
{
rm = new ReadCommand(table, key, columnFamily_column, sinceTimestamp);
}
else
{
rm = new ReadCommand(table, key, columnFamily_column, start, count);
}
rm.setIsDigestQuery(isDigest);
return rm;
String table = dis.readUTF();
String key = dis.readUTF();
String columnFamily_column = dis.readUTF();
int start = dis.readInt();
int count = dis.readInt();
long sinceTimestamp = dis.readLong();
boolean isDigest = dis.readBoolean();
int size = dis.readInt();
List<String> columns = new ArrayList<String>();
for (int i = 0; i < size; ++i)
{
byte[] bytes = new byte[dis.readInt()];
dis.readFully(bytes);
columns.add(new String(bytes));
}
ReadCommand rm = null;
if (columns.size() > 0)
{
rm = new ReadCommand(table, key, columnFamily_column, columns);
}
else if (sinceTimestamp > 0)
{
rm = new ReadCommand(table, key, columnFamily_column, sinceTimestamp);
}
else
{
rm = new ReadCommand(table, key, columnFamily_column, start, count);
}
rm.setDigestQuery(isDigest);
return rm;
}
}

34
src/org/apache/cassandra/db/ReadVerbHandler.java
View File

@ -73,23 +73,23 @@ public class ReadVerbHandler implements IVerbHandler
try
{
ReadCommand readCommand = ReadCommand.serializer().deserialize(readCtx.bufIn_);
Table table = Table.open(readCommand.table());
Table table = Table.open(readCommand.table);
Row row = null;
long start = System.currentTimeMillis();
if( readCommand.columnFamily_column() == null )
row = table.get(readCommand.key());
if( readCommand.columnFamilyColumn == null )
row = table.get(readCommand.key);
else
{
if(readCommand.getColumnNames().size() == 0)
if(readCommand.columnNames.size() == 0)
{
if(readCommand.count() > 0 && readCommand.start() >= 0)
row = table.getRow(readCommand.key(), readCommand.columnFamily_column(), readCommand.start(), readCommand.count());
if(readCommand.count > 0 && readCommand.start >= 0)
row = table.getRow(readCommand.key, readCommand.columnFamilyColumn, readCommand.start, readCommand.count);
else
row = table.getRow(readCommand.key(), readCommand.columnFamily_column());
row = table.getRow(readCommand.key, readCommand.columnFamilyColumn);
}
else
{
row = table.getRow(readCommand.key(), readCommand.columnFamily_column(), readCommand.getColumnNames());
row = table.getRow(readCommand.key, readCommand.columnFamilyColumn, readCommand.columnNames);
}
}
logger_.info("getRow() TIME: " + (System.currentTimeMillis() - start) + " ms.");
@ -121,8 +121,8 @@ public class ReadVerbHandler implements IVerbHandler
logger_.info("ReadVerbHandler TIME 2: " + (System.currentTimeMillis() - start) + " ms.");
/* Do read repair if header of the message says so */
String repair = new String( message.getHeader(ReadCommand.doRepair_) );
if ( repair.equals( ReadCommand.doRepair_ ) )
String repair = new String( message.getHeader(ReadCommand.DO_REPAIR) );
if ( repair.equals( ReadCommand.DO_REPAIR) )
doReadRepair(row, readCommand);
}
catch ( IOException ex)
@ -139,25 +139,25 @@ public class ReadVerbHandler implements IVerbHandler
{
if ( DatabaseDescriptor.getConsistencyCheck() )
{
List<EndPoint> endpoints = StorageService.instance().getNLiveStorageEndPoint(readCommand.key());
List<EndPoint> endpoints = StorageService.instance().getNLiveStorageEndPoint(readCommand.key);
/* Remove the local storage endpoint from the list. */
endpoints.remove( StorageService.getLocalStorageEndPoint() );
if(readCommand.getColumnNames().size() == 0)
if(readCommand.columnNames.size() == 0)
{
if( readCommand.start() >= 0 && readCommand.count() < Integer.MAX_VALUE)
if( readCommand.start >= 0 && readCommand.count < Integer.MAX_VALUE)
{
StorageService.instance().doConsistencyCheck(row, endpoints, readCommand.columnFamily_column(), readCommand.start(), readCommand.count());
StorageService.instance().doConsistencyCheck(row, endpoints, readCommand.columnFamilyColumn, readCommand.start, readCommand.count);
}
if( readCommand.sinceTimestamp() > 0)
if( readCommand.sinceTimestamp > 0)
{
StorageService.instance().doConsistencyCheck(row, endpoints, readCommand.columnFamily_column(), readCommand.sinceTimestamp());
StorageService.instance().doConsistencyCheck(row, endpoints, readCommand.columnFamilyColumn, readCommand.sinceTimestamp);
}
}
else
{
StorageService.instance().doConsistencyCheck(row, endpoints, readCommand.columnFamily_column(), readCommand.getColumnNames());
StorageService.instance().doConsistencyCheck(row, endpoints, readCommand.columnFamilyColumn, readCommand.columnNames);
}
}
}

6
src/org/apache/cassandra/service/ConsistencyManager.java
View File

@ -94,7 +94,7 @@ class ConsistencyManager implements Runnable
String table = DatabaseDescriptor.getTables().get(0);
ReadCommand readCommand = constructReadMessage(false);
// ReadMessage readMessage = new ReadMessage(table, row_.key(), columnFamily_);
Message message = ReadCommand.makeReadMessage(readCommand);
Message message = readCommand.makeReadMessage();
MessagingService.getMessagingInstance().sendRR(message, replicas_.toArray( new EndPoint[0] ), responseHandler);
}
}
@ -186,7 +186,7 @@ class ConsistencyManager implements Runnable
ReadCommand readCommandDigestOnly = constructReadMessage(true);
try
{
Message messageDigestOnly = ReadCommand.makeReadMessage(readCommandDigestOnly);
Message messageDigestOnly = readCommandDigestOnly.makeReadMessage();
IAsyncCallback digestResponseHandler = new DigestResponseHandler();
MessagingService.getMessagingInstance().sendRR(messageDigestOnly, replicas_.toArray(new EndPoint[0]), digestResponseHandler);
}
@ -221,7 +221,7 @@ class ConsistencyManager implements Runnable
readCommand = new ReadCommand(table, row_.key(), columnFamily_, columnNames_);
}
readCommand.setIsDigestQuery(isDigestQuery);
readCommand.setDigestQuery(isDigestQuery);
return readCommand;
}
}

6
src/org/apache/cassandra/service/MultiQuorumResponseHandler.java
View File

@ -130,9 +130,9 @@ public class MultiQuorumResponseHandler implements IAsyncCallback
if ( DatabaseDescriptor.getConsistencyCheck())
{
ReadCommand readCommand = readMessages_.get(key);
readCommand.setIsDigestQuery(false);
Message messageRepair = ReadCommand.makeReadMessage(readCommand);
EndPoint[] endpoints = MultiQuorumResponseHandler.this.endpoints_.get( readCommand.key() );
readCommand.setDigestQuery(false);
Message messageRepair = readCommand.makeReadMessage();
EndPoint[] endpoints = MultiQuorumResponseHandler.this.endpoints_.get(readCommand.key);
Message[][] messages = new Message[][]{ {messageRepair, messageRepair, messageRepair} };
EndPoint[][] epList = new EndPoint[][]{ endpoints };
MessagingService.getMessagingInstance().sendRR(messages, epList, MultiQuorumResponseHandler.this);

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

@ -149,7 +149,7 @@ public class StorageProxy
Set<String> keys = readMessages.keySet();
for ( String key : keys )
{
Message message = ReadCommand.makeReadMessage( readMessages.get(key) );
Message message = readMessages.get(key).makeReadMessage();
messages.put(key, message);
}
return messages;
@ -210,8 +210,8 @@ public class StorageProxy
EndPoint endPoint = StorageService.instance().findSuitableEndPoint(key);
if(endPoint != null)
{
Message message = ReadCommand.makeReadMessage(readCommand);
message.addHeader(ReadCommand.doRepair_, ReadCommand.doRepair_.getBytes());
Message message = readCommand.makeReadMessage();
message.addHeader(ReadCommand.DO_REPAIR, ReadCommand.DO_REPAIR.getBytes());
IAsyncResult iar = MessagingService.getMessagingInstance().sendRR(message, endPoint);
Object[] result = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
byte[] body = (byte[])result[0];
@ -437,7 +437,7 @@ public class StorageProxy
ReadCommand readCommand = new ReadCommand(tablename, key, columnFamily, columns);
ReadCommand readCommandDigestOnly = new ReadCommand(tablename, key, columnFamily, columns);
readCommandDigestOnly.setIsDigestQuery(true);
readCommandDigestOnly.setDigestQuery(true);
Row row = StorageProxy.doStrongReadProtocol(key, readCommand, readCommandDigestOnly);
logger_.debug("readProtocol: " + (System.currentTimeMillis() - startTime) + " ms.");
@ -474,7 +474,7 @@ public class StorageProxy
{
readCommand = new ReadCommand(tablename, key, columnFamily);
}
Message message = ReadCommand.makeReadMessage(readCommand);
Message message = readCommand.makeReadMessage();
if( start >= 0 && count < Integer.MAX_VALUE)
{
readCommandDigestOnly = new ReadCommand(tablename, key, columnFamily, start, count);
@ -483,7 +483,7 @@ public class StorageProxy
{
readCommandDigestOnly = new ReadCommand(tablename, key, columnFamily);
}
readCommandDigestOnly.setIsDigestQuery(true);
readCommandDigestOnly.setDigestQuery(true);
Row row = doStrongReadProtocol(key, readCommand, readCommandDigestOnly);
logger_.debug("readProtocol: " + (System.currentTimeMillis() - startTime) + " ms.");
return row;
@ -508,24 +508,24 @@ public class StorageProxy
Map<String, ReadCommand[]> readMessages = new HashMap<String, ReadCommand[]>();
for (String key : keys )
{
ReadCommand[] readCommand = new ReadCommand[2];
ReadCommand[] readParameters = new ReadCommand[2];
if( start >= 0 && count < Integer.MAX_VALUE)
{
readCommand[0] = new ReadCommand(tablename, key, columnFamily, start, count);
readParameters[0] = new ReadCommand(tablename, key, columnFamily, start, count);
}
else
{
readCommand[0] = new ReadCommand(tablename, key, columnFamily);
readParameters[0] = new ReadCommand(tablename, key, columnFamily);
}
if( start >= 0 && count < Integer.MAX_VALUE)
{
readCommand[1] = new ReadCommand(tablename, key, columnFamily, start, count);
readParameters[1] = new ReadCommand(tablename, key, columnFamily, start, count);
}
else
{
readCommand[1] = new ReadCommand(tablename, key, columnFamily);
readParameters[1] = new ReadCommand(tablename, key, columnFamily);
}
readCommand[1].setIsDigestQuery(true);
readParameters[1].setDigestQuery(true);
}
rows = doStrongReadProtocol(readMessages);
logger_.debug("readProtocol: " + (System.currentTimeMillis() - startTime) + " ms.");
@ -539,9 +539,9 @@ public class StorageProxy
ReadCommand readCommand = null;
ReadCommand readCommandDigestOnly = null;
readCommand = new ReadCommand(tablename, key, columnFamily, sinceTimestamp);
Message message = ReadCommand.makeReadMessage(readCommand);
Message message = readCommand.makeReadMessage();
readCommandDigestOnly = new ReadCommand(tablename, key, columnFamily, sinceTimestamp);
readCommandDigestOnly.setIsDigestQuery(true);
readCommandDigestOnly.setDigestQuery(true);
Row row = doStrongReadProtocol(key, readCommand, readCommandDigestOnly);
logger_.debug("readProtocol: " + (System.currentTimeMillis() - startTime) + " ms.");
return row;
@ -556,8 +556,8 @@ public class StorageProxy
private static Row doStrongReadProtocol(String key, ReadCommand readCommand, ReadCommand readCommandDigest) throws IOException, TimeoutException
{
Row row = null;
Message message = ReadCommand.makeReadMessage(readCommand);
Message messageDigestOnly = ReadCommand.makeReadMessage(readCommandDigest);
Message message = readCommand.makeReadMessage();
Message messageDigestOnly = readCommandDigest.makeReadMessage();
IResponseResolver<Row> readResponseResolver = new ReadResponseResolver();
QuorumResponseHandler<Row> quorumResponseHandler = new QuorumResponseHandler<Row>(
@ -605,9 +605,9 @@ public class StorageProxy
QuorumResponseHandler<Row> quorumResponseHandlerRepair = new QuorumResponseHandler<Row>(
DatabaseDescriptor.getReplicationFactor(),
readResponseResolverRepair);
readCommand.setIsDigestQuery(false);
readCommand.setDigestQuery(false);
logger_.info("DigestMismatchException: " + key);
Message messageRepair = ReadCommand.makeReadMessage(readCommand);
Message messageRepair = readCommand.makeReadMessage();
MessagingService.getMessagingInstance().sendRR(messageRepair, endPoints, quorumResponseHandlerRepair);
try
{
@ -634,11 +634,11 @@ public class StorageProxy
for ( String key : keys )
{
Message[] msg = new Message[DatabaseDescriptor.getReplicationFactor()];
ReadCommand[] readCommand = readMessages.get(key);
msg[0] = ReadCommand.makeReadMessage( readCommand[0] );
ReadCommand[] readParameters = readMessages.get(key);
msg[0] = readParameters[0].makeReadMessage();
for ( int i = 1; i < msg.length; ++i )
{
msg[i] = ReadCommand.makeReadMessage( readCommand[1] );
msg[i] = readParameters[1].makeReadMessage();
}
}
return messages;

2
test/org/apache/cassandra/db/ReadMessageTest.java
View File

@ -25,7 +25,7 @@ public class ReadMessageTest
private ReadCommand serializeAndDeserializeReadMessage(ReadCommand rm)
{
ReadCommand rm2 = null;
ReadCommandSerializer rms = (ReadCommandSerializer) ReadCommand.serializer();
ReadCommandSerializer rms = ReadCommand.serializer();
DataOutputBuffer dos = new DataOutputBuffer();
DataInputBuffer dis = new DataInputBuffer();