diff --git a/src/org/apache/cassandra/db/ReadCommand.java b/src/org/apache/cassandra/db/ReadCommand.java index 85460bef42..d290b26d25 100644 --- a/src/org/apache/cassandra/db/ReadCommand.java +++ b/src/org/apache/cassandra/db/ReadCommand.java @@ -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 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 serializer() + private static List 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 columnNames; + + private boolean isDigestQuery = false; + + public ReadCommand(String table, String key, String columnFamilyColumn, int start, int count, long sinceTimestamp, List 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 columns_ = new ArrayList(); - 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 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 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 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 { - 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 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 columns = new ArrayList(); - 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 columns = new ArrayList(); + 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; } } - diff --git a/src/org/apache/cassandra/db/ReadVerbHandler.java b/src/org/apache/cassandra/db/ReadVerbHandler.java index e4b39bdd1e..4e2185568a 100644 --- a/src/org/apache/cassandra/db/ReadVerbHandler.java +++ b/src/org/apache/cassandra/db/ReadVerbHandler.java @@ -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 endpoints = StorageService.instance().getNLiveStorageEndPoint(readCommand.key()); + List 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); } } } diff --git a/src/org/apache/cassandra/service/ConsistencyManager.java b/src/org/apache/cassandra/service/ConsistencyManager.java index 33238aaa8c..b9ff7d31a0 100644 --- a/src/org/apache/cassandra/service/ConsistencyManager.java +++ b/src/org/apache/cassandra/service/ConsistencyManager.java @@ -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; } } diff --git a/src/org/apache/cassandra/service/MultiQuorumResponseHandler.java b/src/org/apache/cassandra/service/MultiQuorumResponseHandler.java index 2d4dfe3a5a..ea06a721b1 100644 --- a/src/org/apache/cassandra/service/MultiQuorumResponseHandler.java +++ b/src/org/apache/cassandra/service/MultiQuorumResponseHandler.java @@ -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); diff --git a/src/org/apache/cassandra/service/StorageProxy.java b/src/org/apache/cassandra/service/StorageProxy.java index c3d0b66cb5..bea90e5d08 100644 --- a/src/org/apache/cassandra/service/StorageProxy.java +++ b/src/org/apache/cassandra/service/StorageProxy.java @@ -149,7 +149,7 @@ public class StorageProxy Set 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 readMessages = new HashMap(); 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 readResponseResolver = new ReadResponseResolver(); QuorumResponseHandler quorumResponseHandler = new QuorumResponseHandler( @@ -605,9 +605,9 @@ public class StorageProxy QuorumResponseHandler quorumResponseHandlerRepair = new QuorumResponseHandler( 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; diff --git a/test/org/apache/cassandra/db/ReadMessageTest.java b/test/org/apache/cassandra/db/ReadMessageTest.java index f517c9a6dd..d1f1a01718 100644 --- a/test/org/apache/cassandra/db/ReadMessageTest.java +++ b/test/org/apache/cassandra/db/ReadMessageTest.java @@ -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();