From 059e5f166b7956ca3127d4e1c4fd05d0ceeabaf2 Mon Sep 17 00:00:00 2001 From: Eric Evans Date: Wed, 16 Jun 2010 16:16:25 +0000 Subject: [PATCH] working batch_mutate() for avro server w/ tests Patch by eevans git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@955286 13f79535-47bb-0310-9956-ffa450edef68 --- interface/cassandra.avpr | 20 ++-- interface/cassandra.genavro | 10 +- .../apache/cassandra/avro/AvroValidation.java | 3 +- .../cassandra/avro/CassandraServer.java | 30 +++--- test/system/test_avro_server.py | 94 ++++++++++++++++++- 5 files changed, 129 insertions(+), 28 deletions(-) diff --git a/interface/cassandra.avpr b/interface/cassandra.avpr index f840eecb0a..c3b27de7f8 100644 --- a/interface/cassandra.avpr +++ b/interface/cassandra.avpr @@ -99,6 +99,16 @@ {"name": "cf_defs", "type": {"type": "array", "items": "CfDef"}} ] }, + {"name": "MutationsMapEntry", "type": "record", + "fields": [ + {"name": "key", "type" : "bytes"}, + {"name": "mutations", "type": + {"type": "map", + "values": {"type": "array", "items": "Mutation"} + } + } + ] + }, {"name": "ConsistencyLevel", "type": "enum", "symbols": [ "ZERO", "ONE", "QUORUM", "DCQUORUM", "DCQUORUMSYNC", "ALL" @@ -161,14 +171,8 @@ }, "batch_mutate": { "request": [ - {"name": "keyspace", "type": "string"}, - {"name": "mutation_map", - "type": { - "type": "map", "values": { - "type": "map", "values": { - "type": "array", "items": "Mutation"} - } - } + {"name": "mutation_map", "type": + {"type": "array", "items": "MutationsMapEntry"} }, {"name": "consistency_level", "type": "ConsistencyLevel"} ], diff --git a/interface/cassandra.genavro b/interface/cassandra.genavro index 0626403f37..34fcf0a12c 100644 --- a/interface/cassandra.genavro +++ b/interface/cassandra.genavro @@ -82,7 +82,12 @@ protocol Cassandra { string strategy_class; int replication_factor; array cf_defs; - } + } + + record MutationsMapEntry { + bytes key; + map> mutations; + } enum ConsistencyLevel { ZERO, ONE, QUORUM, DCQUORUM, DCQUORUMSYNC, ALL @@ -122,8 +127,7 @@ protocol Cassandra { ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException; - void batch_mutate(string keyspace, - map>> mutation_map, + void batch_mutate(array mutation_map, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException; diff --git a/src/java/org/apache/cassandra/avro/AvroValidation.java b/src/java/org/apache/cassandra/avro/AvroValidation.java index 9f8458ab13..4da386243d 100644 --- a/src/java/org/apache/cassandra/avro/AvroValidation.java +++ b/src/java/org/apache/cassandra/avro/AvroValidation.java @@ -236,7 +236,8 @@ public class AvroValidation { if (del.predicate != null) { - validateSlicePredicate(keyspace, cfName, del.super_column.array(), del.predicate); + byte[] superName = del.super_column == null ? null : del.super_column.array(); + validateSlicePredicate(keyspace, cfName, superName, del.predicate); if (del.predicate.slice_range != null) throw newInvalidRequestException("Deletion does not yet support SliceRange predicates."); } diff --git a/src/java/org/apache/cassandra/avro/CassandraServer.java b/src/java/org/apache/cassandra/avro/CassandraServer.java index 8391edfc32..b8c577c4bf 100644 --- a/src/java/org/apache/cassandra/avro/CassandraServer.java +++ b/src/java/org/apache/cassandra/avro/CassandraServer.java @@ -326,29 +326,28 @@ public class CassandraServer implements Cassandra { } } - public Void batch_mutate(Utf8 keyspace, Map>> mutationMap, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, UnavailableException, TimedOutException + @Override + public Void batch_mutate(GenericArray mutationMap, ConsistencyLevel consistencyLevel) + throws AvroRemoteException, InvalidRequestException, UnavailableException, TimedOutException { if (logger.isDebugEnabled()) logger.debug("batch_mutate"); - String keyspaceString = keyspace.toString(); - List rowMutations = new ArrayList(); - for (Map.Entry>> mutationEntry: mutationMap.entrySet()) + + for (MutationsMapEntry pair: mutationMap) { - String key = mutationEntry.getKey().toString(); - AvroValidation.validateKey(key); + AvroValidation.validateKey(pair.key.array()); + Map> cfToMutations = pair.mutations; - Map> cfToMutations = mutationEntry.getValue(); for (Map.Entry> cfMutations : cfToMutations.entrySet()) { String cfName = cfMutations.getKey().toString(); for (Mutation mutation : cfMutations.getValue()) - AvroValidation.validateMutation(keyspaceString, cfName, mutation); + AvroValidation.validateMutation(curKeyspace.get(), cfName, mutation); } - rowMutations.add(getRowMutationFromMutations(keyspaceString, key, cfToMutations)); + rowMutations.add(getRowMutationFromMutations(curKeyspace.get(), pair.key.array(), cfToMutations)); } if (consistencyLevel == ConsistencyLevel.ZERO) @@ -381,10 +380,9 @@ public class CassandraServer implements Cassandra { } // FIXME: This is copypasta from o.a.c.db.RowMutation, (RowMutation.getRowMutation uses Thrift types directly). - private static RowMutation getRowMutationFromMutations(String keyspace, String key, Map> cfMap) + private static RowMutation getRowMutationFromMutations(String keyspace, byte[] key, Map> cfMap) { - // FIXME: string key - RowMutation rm = new RowMutation(keyspace, key.trim().getBytes(UTF8)); + RowMutation rm = new RowMutation(keyspace, key); for (Map.Entry> entry : cfMap.entrySet()) { @@ -419,6 +417,8 @@ public class CassandraServer implements Cassandra { // FIXME: This is copypasta from o.a.c.db.RowMutation, (RowMutation.getRowMutation uses Thrift types directly). private static void deleteColumnOrSuperColumnToRowMutation(RowMutation rm, String cfName, Deletion del) { + byte[] superName = del.super_column == null ? null : del.super_column.array(); + if (del.predicate != null && del.predicate.column_names != null) { for (ByteBuffer col : del.predicate.column_names) @@ -426,12 +426,12 @@ public class CassandraServer implements Cassandra { if (del.super_column == null && DatabaseDescriptor.getColumnFamilyType(rm.getTable(), cfName) == ColumnFamilyType.Super) rm.delete(new QueryPath(cfName, col.array()), unavronateClock(del.clock)); else - rm.delete(new QueryPath(cfName, del.super_column.array(), col.array()), unavronateClock(del.clock)); + rm.delete(new QueryPath(cfName, superName, col.array()), unavronateClock(del.clock)); } } else { - rm.delete(new QueryPath(cfName, del.super_column.array()), unavronateClock(del.clock)); + rm.delete(new QueryPath(cfName, superName), unavronateClock(del.clock)); } } diff --git a/test/system/test_avro_server.py b/test/system/test_avro_server.py index 07c8509a30..b20de988b3 100644 --- a/test/system/test_avro_server.py +++ b/test/system/test_avro_server.py @@ -23,6 +23,18 @@ import struct def i64(i): return struct.pack('>q', i) +def timestamp(): + return long(time() * 1e6) + +def new_column(suffix, stamp=None, ttl=0): + ts = isinstance(stamp, (long,int)) and stamp or timestamp() + column = dict() + column['name'] = 'name-%s' % suffix + column['value'] = 'value-%s' % suffix + column['clock'] = {'timestamp': ts} + column['ttl'] = ttl + return column + def assert_columns_match(colA, colB): assert colA['name'] == colB['name'], \ "column name mismatch: %s != %s" % (colA['name'], colB['name']) @@ -33,7 +45,12 @@ def assert_cosc(thing, with_supercolumn=False): containing = with_supercolumn and 'super_column' or 'column' assert isinstance(thing, dict), "Expected dict, got %s" % type(thing) assert thing.has_key(containing) and thing[containing].has_key('name'), \ - "Invalid or missing \"%s\"" % containing + "Invalid or missing \"%s\" member" % containing + +def assert_raises(excClass, func, *args, **kwargs): + try: r = func(*args, **kwargs) + except excClass: pass + else: raise Exception('expected %s; got %s' % (excClass.__name__, r)) class TestRpcOperations(AvroTester): def test_insert_simple(self): # Also tests get @@ -125,6 +142,58 @@ class TestRpcOperations(AvroTester): except AvroRemoteException, err: pass else: assert False, "Expected exception, returned %s instead" % cosc + def test_batch_mutate(self): + "batching addition/removal mutations" + self.__set_keyspace('Keyspace1') + + mutations = list() + + # New column mutations + for i in range(3): + cosc = {'column': new_column(i)} + mutation = {'column_or_supercolumn': cosc} + mutations.append(mutation) + + map_entry = {'key': 'key1', 'mutations': {'Standard1': mutations}} + + params = dict() + params['mutation_map'] = [map_entry] + params['consistency_level'] = 'ONE' + + self.client.request('batch_mutate', params) + + # Verify that new columns were added + for i in range(3): + column = new_column(i) + cosc = self.__get('key1', 'Standard1', None, column['name']) + assert_cosc(cosc) + assert_columns_match(cosc['column'], column) + + # Add one more column; remove one column + extra_column = new_column(3); remove_column = new_column(0) + mutations = [{'column_or_supercolumn': {'column': extra_column}}] + deletion = dict() + deletion['clock'] = {'timestamp': timestamp()} + deletion['predicate'] = {'column_names': [remove_column['name']]} + mutations.append({'deletion': deletion}) + + map_entry = {'key': 'key1', 'mutations': {'Standard1': mutations}} + + params = dict() + params['mutation_map'] = [map_entry] + params['consistency_level'] = 'ONE' + + self.client.request('batch_mutate', params) + + # Ensure successful column removal + assert_raises(AvroRemoteException, + self.__get, 'key1', 'Standard1', None, remove_column['name']) + + # Ensure successful column addition + cosc = self.__get('key1', 'Standard1', None, extra_column['name']) + assert_cosc(cosc) + assert_columns_match(cosc['column'], extra_column) + def test_describe_keyspaces(self): "retrieving a list of all keyspaces" keyspaces = self.client.request('describe_keyspaces', {}) @@ -143,4 +212,27 @@ class TestRpcOperations(AvroTester): assert len(segs) == 3 and len([i for i in segs if i.isdigit()]) == 3, \ "incorrect api version format: " + vers + def __get(self, key, cf, super_name, col_name, consistency_level='ONE'): + """ + Given arguments for the key, column family, super column name, + column name, and consistency level, returns a dictionary + representing a ColumnOrSuperColumn record. + + Raises an AvroRemoteException if the column is not found. + """ + params = dict() + params['key'] = key + params['column_path'] = dict() + params['column_path']['column_family'] = cf + params['column_path']['column'] = col_name + params['consistency_level'] = consistency_level + + if (super_name): + params['super_column'] = super_name + + return self.client.request('get', params) + + def __set_keyspace(self, keyspace_name): + self.client.request('set_keyspace', {'keyspace': keyspace_name}) + # vi:ai sw=4 ts=4 tw=0 et