From 41598bfead845723d5eaaacc0cf4f9372825f3a4 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 27 Mar 2009 02:43:53 +0000 Subject: [PATCH] move row mutation factory code into RowMutation; change RM.add(name, cf) to RM.add(cf) git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@758993 13f79535-47bb-0310-9956-ffa450edef68 --- src/org/apache/cassandra/db/RowMutation.java | 179 ++++++++------ .../cassandra/service/CassandraServer.java | 226 ++++-------------- .../service/ReadResponseResolver.java | 2 +- .../cassandra/db/ColumnFamilyStoreTest.java | 4 +- 4 files changed, 153 insertions(+), 258 deletions(-) diff --git a/src/org/apache/cassandra/db/RowMutation.java b/src/org/apache/cassandra/db/RowMutation.java index 616598c842..9cfcb14b48 100644 --- a/src/org/apache/cassandra/db/RowMutation.java +++ b/src/org/apache/cassandra/db/RowMutation.java @@ -18,19 +18,29 @@ package org.apache.cassandra.db; -import java.util.*; +import java.io.ByteArrayOutputStream; import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; import java.io.Serializable; -import java.io.ByteArrayOutputStream; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; import org.apache.commons.lang.ArrayUtils; +import org.apache.commons.lang.StringUtils; import org.apache.cassandra.io.ICompactSerializer; -import org.apache.cassandra.net.Message; import org.apache.cassandra.net.EndPoint; +import org.apache.cassandra.net.Message; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.batch_mutation_super_t; +import org.apache.cassandra.service.batch_mutation_t; +import org.apache.cassandra.service.column_t; +import org.apache.cassandra.service.superColumn_t; import org.apache.cassandra.utils.FBUtilities; @@ -40,13 +50,13 @@ import org.apache.cassandra.utils.FBUtilities; public class RowMutation implements Serializable { - private static ICompactSerializer serializer_; + private static ICompactSerializer serializer_; public static final String HINT = "HINT"; static { serializer_ = new RowMutationSerializer(); - } + } static ICompactSerializer serializer() { @@ -56,7 +66,6 @@ public class RowMutation implements Serializable private String table_; private String key_; protected Map modifications_ = new HashMap(); - protected Map deletions_ = new HashMap(); /* Ctor for JAXB */ private RowMutation() @@ -73,20 +82,17 @@ public class RowMutation implements Serializable { table_ = table; key_ = row.key(); - Map cfSet = row.getColumnFamilyMap(); - Set keyset = cfSet.keySet(); - for(String cfName : keyset) + for (ColumnFamily cf : row.getColumnFamilies()) { - add(cfName, cfSet.get(cfName)); + add(cf); } } - protected RowMutation(String table, String key, Map modifications, Map deletions) + protected RowMutation(String table, String key, Map modifications) { - table_ = table; - key_ = key; - modifications_ = modifications; - deletions_ = deletions; + table_ = table; + key_ = key; + modifications_ = modifications; } public static String[] getColumnAndColumnFamily(String cf) @@ -107,7 +113,7 @@ public class RowMutation implements Serializable void addHints(String hint) throws IOException, ColumnFamilyNotDefinedException { String cfName = Table.hints_ + ":" + hint; - add(cfName, new byte[0]); + add(cfName, ArrayUtils.EMPTY_BYTE_ARRAY, 0); } /* @@ -116,24 +122,13 @@ public class RowMutation implements Serializable * param @ cf - column family name * param @ columnFamily - the column family. */ - public void add(String cf, ColumnFamily columnFamily) + public void add(ColumnFamily columnFamily) { - modifications_.put(cf, columnFamily); - } - - /* - * Specify a column name and a corresponding value for - * the column. Column name is specified as :column. - * This will result in a ColumnFamily associated with - * as name and a Column with - * as name. - * - * param @ cf - column name as : - * param @ value - value associated with the column - */ - public void add(String cf, byte[] value) throws IOException, ColumnFamilyNotDefinedException - { - add(cf, value, 0); + if (modifications_.containsKey(columnFamily.name())) + { + throw new IllegalArgumentException("ColumnFamily " + columnFamily.name() + " is already being modified"); + } + modifications_.put(columnFamily.name(), columnFamily); } /* @@ -206,7 +201,6 @@ public class RowMutation implements Serializable modifications_.put(cfName, columnFamily); } - /* * This is equivalent to calling commit. Applies the changes to * to the table that is obtained by calling Table.open(). @@ -237,15 +231,15 @@ public class RowMutation implements Serializable * This is equivalent to calling commit. Applies the changes to * to the table that is obtained by calling Table.open(). */ - void load(Row row) throws IOException, ColumnFamilyNotDefinedException + void load(Row row) throws IOException, ColumnFamilyNotDefinedException, ExecutionException, InterruptedException { Table table = Table.open(table_); Set cfNames = modifications_.keySet(); - for (String cfName : cfNames ) + for (String cfName : cfNames) { - if ( !table.isValidColumnFamily(cfName) ) + if (!table.isValidColumnFamily(cfName)) throw new ColumnFamilyNotDefinedException("Column Family " + cfName + " has not been defined."); - row.addColumnFamily( modifications_.get(cfName) ); + row.addColumnFamily(modifications_.get(cfName)); } table.load(row); } @@ -264,65 +258,110 @@ public class RowMutation implements Serializable EndPoint from = (local != null) ? local : new EndPoint(FBUtilities.getHostName(), 7000); return new Message(from, StorageService.mutationStage_, verbHandlerName, bos.toByteArray()); } + + public static RowMutation getRowMutation(batch_mutation_t batchMutation) + { + RowMutation rm = new RowMutation(batchMutation.table, + batchMutation.key.trim()); + for (String cfname : batchMutation.cfmap.keySet()) + { + List list = batchMutation.cfmap.get(cfname); + for (column_t columnData : list) + { + rm.add(cfname + ":" + columnData.columnName, + columnData.value.getBytes(), columnData.timestamp); + + } + } + return rm; + } + + public static RowMutation getRowMutation(batch_mutation_super_t batchMutationSuper) + { + RowMutation rm = new RowMutation(batchMutationSuper.table, + batchMutationSuper.key.trim()); + Set keys = batchMutationSuper.cfmap.keySet(); + Iterator keyIter = keys.iterator(); + while (keyIter.hasNext()) + { + Object key = keyIter.next(); // Get the next key. + List list = batchMutationSuper.cfmap.get(key); + for (superColumn_t superColumnData : list) + { + if (superColumnData.columns.size() != 0) + { + for (column_t columnData : superColumnData.columns) + { + rm.add(key.toString() + ":" + superColumnData.name + ":" + columnData.columnName, + columnData.value.getBytes(), columnData.timestamp); + } + } + else + { + rm.add(key.toString() + ":" + superColumnData.name, ArrayUtils.EMPTY_BYTE_ARRAY, 0); + } + } + } + return rm; + } + + public String toString() + { + return "RowMutation(" + + "key='" + key_ + '\'' + + ", modifications=[" + StringUtils.join(modifications_.values(), ", ") + "]" + + ')'; + } } class RowMutationSerializer implements ICompactSerializer { - private void freezeTheMaps(Map map, DataOutputStream dos) throws IOException - { - int size = map.size(); + private void freezeTheMaps(Map map, DataOutputStream dos) throws IOException + { + int size = map.size(); dos.writeInt(size); - if ( size > 0 ) + if (size > 0) { Set keys = map.keySet(); - for( String key : keys ) + for (String key : keys) { - dos.writeUTF(key); + dos.writeUTF(key); ColumnFamily cf = map.get(key); - if ( cf != null ) + if (cf != null) { ColumnFamily.serializer().serialize(cf, dos); } } } - } + } - public void serialize(RowMutation rm, DataOutputStream dos) throws IOException - { - dos.writeUTF(rm.table()); - dos.writeUTF(rm.key()); + public void serialize(RowMutation rm, DataOutputStream dos) throws IOException + { + dos.writeUTF(rm.table()); + dos.writeUTF(rm.key()); - /* serialize the modifications_ in the mutation */ + /* serialize the modifications_ in the mutation */ freezeTheMaps(rm.modifications_, dos); + } - /* serialize the deletions_ in the mutation */ - freezeTheMaps(rm.deletions_, dos); - } - - private Map defreezeTheMaps(DataInputStream dis) throws IOException - { - Map map = new HashMap(); + private Map defreezeTheMaps(DataInputStream dis) throws IOException + { + Map map = new HashMap(); int size = dis.readInt(); - for ( int i = 0; i < size; ++i ) + for (int i = 0; i < size; ++i) { String key = dis.readUTF(); ColumnFamily cf = ColumnFamily.serializer().deserialize(dis); map.put(key, cf); } return map; - } + } public RowMutation deserialize(DataInputStream dis) throws IOException { - String table = dis.readUTF(); - String key = dis.readUTF(); - - /* Defreeze the modifications_ map */ - Map modifications = defreezeTheMaps(dis); - - /* Defreeze the deletions_ map */ - Map deletions = defreezeTheMaps(dis); - - return new RowMutation(table, key, modifications, deletions); + String table = dis.readUTF(); + String key = dis.readUTF(); + Map modifications = defreezeTheMaps(dis); + return new RowMutation(table, key, modifications); } } diff --git a/src/org/apache/cassandra/service/CassandraServer.java b/src/org/apache/cassandra/service/CassandraServer.java index 9e5d6bbef9..8837f2fac7 100644 --- a/src/org/apache/cassandra/service/CassandraServer.java +++ b/src/org/apache/cassandra/service/CassandraServer.java @@ -18,21 +18,29 @@ package org.apache.cassandra.service; -import com.facebook.thrift.*; -import com.facebook.thrift.server.*; -import com.facebook.thrift.server.TThreadPoolServer.Options; -import com.facebook.thrift.transport.*; -import com.facebook.thrift.protocol.*; -import com.facebook.fb303.FacebookBase; -import com.facebook.fb303.fb_status; -import java.io.*; -import java.util.Collection; -import java.util.List; +import java.io.BufferedInputStream; +import java.io.FileInputStream; +import java.io.IOException; import java.util.ArrayList; +import java.util.Collection; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Set; +import java.util.Arrays; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.TimeUnit; +import org.apache.log4j.Logger; + +import com.facebook.fb303.FacebookBase; +import com.facebook.fb303.fb_status; +import com.facebook.thrift.TException; +import com.facebook.thrift.protocol.TBinaryProtocol; +import com.facebook.thrift.protocol.TProtocolFactory; +import com.facebook.thrift.server.TThreadPoolServer; +import com.facebook.thrift.server.TThreadPoolServer.Options; +import com.facebook.thrift.transport.TServerSocket; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql.common.CqlResult; @@ -45,8 +53,11 @@ import org.apache.cassandra.db.RowMutationMessage; import org.apache.cassandra.net.EndPoint; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.net.IAsyncResult; import org.apache.cassandra.utils.LogUtil; -import org.apache.log4j.Logger; +import org.apache.cassandra.io.DataInputBuffer; +import org.apache.cassandra.io.DataOutputBuffer; + /** * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com ) */ @@ -69,7 +80,7 @@ public class CassandraServer extends FacebookBase implements storageService = StorageService.instance(); } - public CassandraServer() throws Throwable + public CassandraServer() { super("CassandraServer"); // Create the instance of the storage service @@ -81,7 +92,7 @@ public class CassandraServer extends FacebookBase implements * specified port. */ public void start() throws Throwable - { + { LogUtil.init(); //LogUtil.setLogLevel("com.facebook", "DEBUG"); // Start the storage service @@ -474,107 +485,21 @@ public class CassandraServer extends FacebookBase implements public boolean batch_insert_blocking(batch_mutation_t batchMutation) { - // 1. Get the N nodes from storage service where the data needs to be - // replicated - // 2. Construct a message for read\write - // 3. SendRR ( to all the nodes above ) - // 4. Wait for a response from atleast X nodes where X <= N - // 5. return success - boolean result = false; - try - { - logger_.warn(" batch_insert_blocking"); - validateTable(batchMutation.table); - IResponseResolver writeResponseResolver = new WriteResponseResolver(); - QuorumResponseHandler quorumResponseHandler = new QuorumResponseHandler( - DatabaseDescriptor.getReplicationFactor(), - writeResponseResolver); - EndPoint[] endpoints = storageService.getNStorageEndPoint(batchMutation.key); - // TODO: throw a thrift exception if we do not have N nodes - - logger_.debug(" Creating the row mutation"); - RowMutation rm = new RowMutation(batchMutation.table, - batchMutation.key.trim()); - Set keys = batchMutation.cfmap.keySet(); - Iterator keyIter = keys.iterator(); - while (keyIter.hasNext()) - { - Object key = keyIter.next(); // Get the next key. - List list = batchMutation.cfmap.get(key); - for (column_t columnData : list) - { - rm.add(key.toString() + ":" + columnData.columnName, - columnData.value.getBytes(), columnData.timestamp); - - } - } - - RowMutationMessage rmMsg = new RowMutationMessage(rm); - Message message = new Message(StorageService.getLocalStorageEndPoint(), - StorageService.mutationStage_, - StorageService.mutationVerbHandler_, - new Object[]{ rmMsg } - ); - MessagingService.getMessagingInstance().sendRR(message, endpoints, - quorumResponseHandler); - logger_.debug(" Calling quorum response handler's get"); - result = quorumResponseHandler.get(); - - // TODO: if the result is false that means the writes to all the - // servers failed hence we need to throw an exception or return an - // error back to the client so that it can take appropriate action. - } - catch (Exception e) - { - logger_.info( LogUtil.throwableToString(e) ); - } - return result; - + logger_.debug("batch_insert_blocking"); + RowMutation rm = RowMutation.getRowMutation(batchMutation); + return StorageProxy.insertBlocking(rm); } + public void batch_insert(batch_mutation_t batchMutation) { - // 1. Get the N nodes from storage service where the data needs to be - // replicated - // 2. Construct a message for read\write - // 3. SendRR ( to all the nodes above ) - // 4. Wait for a response from atleast X nodes where X <= N - // 5. return success - - try - { - logger_.debug(" batch_insert"); - logger_.debug(" Creating the row mutation"); - validateTable(batchMutation.table); - RowMutation rm = new RowMutation(batchMutation.table, - batchMutation.key.trim()); - if(batchMutation.cfmap != null) - { - Set keys = batchMutation.cfmap.keySet(); - Iterator keyIter = keys.iterator(); - while (keyIter.hasNext()) - { - Object key = keyIter.next(); // Get the next key. - List list = batchMutation.cfmap.get(key); - for (column_t columnData : list) - { - rm.add(key.toString() + ":" + columnData.columnName, - columnData.value.getBytes(), columnData.timestamp); - - } - } - } - StorageProxy.insert(rm); - } - catch (Exception e) - { - logger_.info( LogUtil.throwableToString(e) ); - } - return; + logger_.debug("batch_insert"); + RowMutation rm = RowMutation.getRowMutation(batchMutation); + StorageProxy.insert(rm); } public void remove(String tablename, String key, String columnFamily_column) { - throw new UnsupportedOperationException(); + throw new UnsupportedOperationException("Remove is coming soon"); } public boolean remove(String tablename, String key, String columnFamily_column, long timestamp, int block_for) @@ -795,86 +720,16 @@ public class CassandraServer extends FacebookBase implements public boolean batch_insert_superColumn_blocking(batch_mutation_super_t batchMutationSuper) { - boolean result = false; - try - { - logger_.warn(" batch_insert_SuperColumn_blocking"); - logger_.debug(" Creating the row mutation"); - validateTable(batchMutationSuper.table); - RowMutation rm = new RowMutation(batchMutationSuper.table, - batchMutationSuper.key.trim()); - Set keys = batchMutationSuper.cfmap.keySet(); - Iterator keyIter = keys.iterator(); - while (keyIter.hasNext()) - { - Object key = keyIter.next(); // Get the next key. - List list = batchMutationSuper.cfmap.get(key); - for (superColumn_t superColumnData : list) - { - if(superColumnData.columns.size() != 0 ) - { - for (column_t columnData : superColumnData.columns) - { - rm.add(key.toString() + ":" + superColumnData.name +":" + columnData.columnName, - columnData.value.getBytes(), columnData.timestamp); - } - } - else - { - rm.add(key.toString() + ":" + superColumnData.name, new byte[0], 0); - } - } - } - StorageProxy.insert(rm); - } - catch (Exception e) - { - logger_.info( LogUtil.throwableToString(e) ); - } - return result; - + logger_.debug("batch_insert_SuperColumn_blocking"); + RowMutation rm = RowMutation.getRowMutation(batchMutationSuper); + return StorageProxy.insertBlocking(rm); } + public void batch_insert_superColumn(batch_mutation_super_t batchMutationSuper) { - try - { - logger_.debug(" batch_insert"); - logger_.debug(" Creating the row mutation"); - validateTable(batchMutationSuper.table); - RowMutation rm = new RowMutation(batchMutationSuper.table, - batchMutationSuper.key.trim()); - if(batchMutationSuper.cfmap != null) - { - Set keys = batchMutationSuper.cfmap.keySet(); - Iterator keyIter = keys.iterator(); - while (keyIter.hasNext()) - { - Object key = keyIter.next(); // Get the next key. - List list = batchMutationSuper.cfmap.get(key); - for (superColumn_t superColumnData : list) - { - if(superColumnData.columns.size() != 0 ) - { - for (column_t columnData : superColumnData.columns) - { - rm.add(key.toString() + ":" + superColumnData.name +":" + columnData.columnName, - columnData.value.getBytes(), columnData.timestamp); - } - } - else - { - rm.add(key.toString() + ":" + superColumnData.name, new byte[0], 0); - } - } - } - } - StorageProxy.insert(rm); - } - catch (Exception e) - { - logger_.info( LogUtil.throwableToString(e) ); - } - return; + logger_.debug("batch_insert_SuperColumn"); + RowMutation rm = RowMutation.getRowMutation(batchMutationSuper); + StorageProxy.insert(rm); } public String getStringProperty(String propertyName) throws TException @@ -961,7 +816,7 @@ public class CassandraServer extends FacebookBase implements } return result; } - + /* * This method is used to ensure that all keys * prior to the specified key, as dtermined by @@ -999,6 +854,7 @@ public class CassandraServer extends FacebookBase implements public static void main(String[] args) throws Throwable { int port = 9160; + try { CassandraServer peerStorageServer = new CassandraServer(); diff --git a/src/org/apache/cassandra/service/ReadResponseResolver.java b/src/org/apache/cassandra/service/ReadResponseResolver.java index d8d3023516..8969956b33 100644 --- a/src/org/apache/cassandra/service/ReadResponseResolver.java +++ b/src/org/apache/cassandra/service/ReadResponseResolver.java @@ -147,7 +147,7 @@ public class ReadResponseResolver implements IResponseResolver for ( String cfName : cfNames ) { ColumnFamily cf = columnFamilies.get(cfName); - rowMutation.add(cfName, cf); + rowMutation.add(cf); } RowMutationMessage rowMutationMessage = new RowMutationMessage(rowMutation); // schedule the read repair diff --git a/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java index 8abadbbcf8..a2882cbb8e 100644 --- a/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -73,7 +73,7 @@ public class ColumnFamilyStoreTest extends ServerTest { rm = new RowMutation("Table1", "key1"); ColumnFamily cf = new ColumnFamily("Standard1"); cf.delete(1); - rm.add(cf.name(), cf); + rm.add(cf); rm.apply(); ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1", new IdentityFilter()); @@ -98,7 +98,7 @@ public class ColumnFamilyStoreTest extends ServerTest { SuperColumn sc = new SuperColumn("SC1"); sc.markForDeleteAt(1); cf.addColumn(sc); - rm.add(cf.name(), cf); + rm.add(cf); rm.apply(); List families = store.getColumnFamilies("key1", "Super1", new IdentityFilter());