diff --git a/contrib/py_stress/avro_stress.py b/contrib/py_stress/avro_stress.py deleted file mode 100644 index 6c0cb9dece..0000000000 --- a/contrib/py_stress/avro_stress.py +++ /dev/null @@ -1,376 +0,0 @@ -#!/usr/bin/python -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# expects a Cassandra server to be running and listening on port 9160. -# (read tests expect insert tests to have run first too.) - -have_multiproc = False -try: - from multiprocessing import Array as array, Process as Thread - from uuid import uuid1 as get_ident - Thread.isAlive = Thread.is_alive - have_multiproc = True -except ImportError: - from threading import Thread - from thread import get_ident - from array import array -from hashlib import md5 -import time, random, sys, os -from random import randint, gauss -from optparse import OptionParser - -import avro.ipc as ipc -import avro.protocol as protocol -from avro.ipc import AvroRemoteException - -L = os.path.abspath(__file__).split(os.path.sep)[:-3] -root = os.path.sep.join(L) - - -parser = OptionParser() -parser.add_option('-n', '--num-keys', type="int", dest="numkeys", - help="Number of keys", default=1000**2) -parser.add_option('-t', '--threads', type="int", dest="threads", - help="Number of threads/procs to use", default=50) -parser.add_option('-c', '--columns', type="int", dest="columns", - help="Number of columns per key", default=5) -parser.add_option('-d', '--nodes', type="string", dest="nodes", - help="Host nodes (comma separated)", default="localhost") -parser.add_option('-s', '--stdev', type="float", dest="stdev", default=0.1, - help="standard deviation factor") -parser.add_option('-r', '--random', action="store_true", dest="random", - help="use random key generator (stdev will have no effect)") -parser.add_option('-f', '--file', type="string", dest="file", - help="write output to file") -parser.add_option('-p', '--port', type="int", default=9160, dest="port", - help="thrift port") -parser.add_option('-m', '--unframed', action="store_true", dest="unframed", - help="use unframed transport") -parser.add_option('-o', '--operation', type="choice", dest="operation", - default="insert", choices=('insert', 'read', 'rangeslice'), - help="operation to perform") -parser.add_option('-u', '--supercolumns', type="int", dest="supers", default=1, - help="number of super columns per key") -parser.add_option('-y', '--family-type', type="choice", dest="cftype", - choices=('regular','super'), default='regular', - help="column family type") -parser.add_option('-k', '--keep-going', action="store_true", dest="ignore", - help="ignore errors inserting or reading") -parser.add_option('-i', '--progress-interval', type="int", default=10, - dest="interval", help="progress report interval (seconds)") -parser.add_option('-g', '--get-range-slice-count', type="int", default=1000, - dest="rangecount", - help="amount of keys to get_range_slices per call") -parser.add_option('-l', '--replication-factor', type="int", default=1, - dest="replication", - help="replication factor to use when creating needed column families") -parser.add_option('-e', '--consistency-level', type="str", default='ONE', - dest="consistency", help="consistency level to use") - -(options, args) = parser.parse_args() - -total_keys = options.numkeys -n_threads = options.threads -keys_per_thread = total_keys / n_threads -columns_per_key = options.columns -supers_per_key = options.supers -# this allows client to round robin requests directly for -# simple request load-balancing -nodes = options.nodes.split(',') - -# a generator that generates all keys according to a bell curve centered -# around the middle of the keys generated (0..total_keys). Remember that -# about 68% of keys will be within stdev away from the mean and -# about 95% within 2*stdev. -stdev = total_keys * options.stdev -mean = total_keys / 2 - -c_levels = ['ZERO', 'ANY', 'ONE', 'QUORUM', 'DCQUORUM', 'DCQUORUMSYNC', 'ALL'] -consistency = options.consistency -if not consistency in c_levels: - print "%s is not a valid consistency level" % options.consistency - sys.exit(3) - -def key_generator_gauss(): - fmt = '%0' + str(len(str(total_keys))) + 'd' - while True: - guess = gauss(mean, stdev) - if 0 <= guess < total_keys: - return fmt % int(guess) - -# a generator that will generate all keys w/ equal probability. this is the -# worst case for caching. -def key_generator_random(): - fmt = '%0' + str(len(str(total_keys))) + 'd' - return fmt % randint(0, total_keys - 1) - -key_generator = key_generator_gauss -if options.random: - key_generator = key_generator_random - -def get_client(host='127.0.0.1', port=9170): - schema = os.path.join(root, 'interface/avro', 'cassandra.avpr') - proto = protocol.parse(open(schema).read()) - client = ipc.HTTPTransceiver(host, port) - return ipc.Requestor(proto, client) - -def make_keyspaces(): - keyspace1 = dict() - keyspace1['name'] = 'Keyspace1' - keyspace1['replication_factor'] = options.replication - keyspace1['strategy_class'] = 'org.apache.cassandra.locator.SimpleStrategy' - - keyspace1['cf_defs'] = [{ - 'keyspace': 'Keyspace1', - 'name': 'Standard1', - }] - - keyspace1['cf_defs'].append({ - 'keyspace': 'Keyspace1', - 'name': 'Super1', - 'column_type': 'Super', - 'comparator_type': 'BytesType', - 'subcomparator_type': 'BytesType', - }) - client = get_client(nodes[0], options.port) - try: - client.request('system_add_keyspace', {'ks_def': keyspace1}) - except AvroRemoteException, e: - print e - client.transceiver.conn.close() - -class Operation(Thread): - def __init__(self, i, opcounts, keycounts, latencies): - Thread.__init__(self) - # generator of the keys to be used - self.range = xrange(keys_per_thread * i, keys_per_thread * (i + 1)) - # we can't use a local counter, since that won't be visible to the parent - # under multiprocessing. instead, the parent passes a "opcounts" array - # and an index that is our assigned counter. - self.idx = i - self.opcounts = opcounts - # similarly, a shared array for latency and key totals - self.latencies = latencies - self.keycounts = keycounts - # random host for pseudo-load-balancing - [hostname] = random.sample(nodes, 1) - # open client - self.cclient = get_client(hostname, options.port) - self.cclient.request('set_keyspace', {'keyspace': 'Keyspace1'}) - -class Inserter(Operation): - def run(self): - data = md5(str(get_ident())).hexdigest() - columns = [{'name': 'C' + str(j), 'value': data, 'timestamp': int(time.time() * 1000000)} for j in xrange(columns_per_key)] - fmt = '%0' + str(len(str(total_keys))) + 'd' - if 'super' == options.cftype: - supers = [{'name': 'S' + str(j), 'columns': columns} for j in xrange(supers_per_key)] - for i in self.range: - key = fmt % i - if 'super' == options.cftype: - cfmap= {'key': key, 'mutations': {'Super1' : [{'column_or_supercolumn': {'super_column': s}} for s in supers]}} - else: - cfmap = {'key': key, 'mutations': {'Standard1': [{'column_or_supercolumn': {'column': c}} for c in columns]}} - start = time.time() - try: - self.cclient.request('batch_mutate', {'mutation_map': [cfmap], 'consistency_level': consistency}) - except KeyboardInterrupt: - raise - except Exception, e: - if options.ignore: - print e - else: - raise - self.latencies[self.idx] += time.time() - start - self.opcounts[self.idx] += 1 - self.keycounts[self.idx] += 1 - - -class Reader(Operation): - def run(self): - p = {'slice_range': {'start': '', 'finish': '', 'reversed': False, 'count': columns_per_key}} - if 'super' == options.cftype: - for i in xrange(keys_per_thread): - key = key_generator() - for j in xrange(supers_per_key): - parent = {'column_family': 'Super1', 'super_column': 'S' + str(j)} - start = time.time() - try: - r = self.cclient.request('get_slice', {'key': key, 'column_parent': parent, 'predicate': p, 'consistency_level': consistency}) - if not r: raise RuntimeError("Key %s not found" % key) - except KeyboardInterrupt: - raise - except Exception, e: - if options.ignore: - print e - else: - raise - self.latencies[self.idx] += time.time() - start - self.opcounts[self.idx] += 1 - self.keycounts[self.idx] += 1 - else: - parent = {'column_family': 'Standard1'} - for i in xrange(keys_per_thread): - key = key_generator() - start = time.time() - try: - r = self.cclient.request('get_slice', {'key': key, 'column_parent': parent, 'predicate': p, 'consistency_level': consistency}) - if not r: raise RuntimeError("Key %s not found" % key) - except KeyboardInterrupt: - raise - except Exception, e: - if options.ignore: - print e - else: - raise - self.latencies[self.idx] += time.time() - start - self.opcounts[self.idx] += 1 - self.keycounts[self.idx] += 1 - -class RangeSlicer(Operation): - def run(self): - begin = self.range[0] - end = self.range[-1] - current = begin - last = current + options.rangecount - fmt = '%0' + str(len(str(total_keys))) + 'd' - p = {'slice_range': {'start': '', 'finish': '', 'reversed': False, 'count': columns_per_key}} - if 'super' == options.cftype: - while current < end: - keyrange = {'start_key': fmt % current, 'end_key': fmt % last, 'count': options.rangecount} - res = [] - for j in xrange(supers_per_key): - parent = {'column_family': 'Super1', 'super_column': 'S' + str(j)} - begin = time.time() - try: - res = self.cclient.request('get_range_slices', {'column_parent': parent, 'predicate': p, 'range': keyrange, 'consistency_level': consistency}) - if not res: raise RuntimeError("Key %s not found" % key) - except KeyboardInterrupt: - raise - except Exception, e: - if options.ignore: - print e - else: - raise - self.latencies[self.idx] += time.time() - begin - self.opcounts[self.idx] += 1 - current += len(r) + 1 - last = current + len(r) + 1 - self.keycounts[self.idx] += len(r) - else: - parent = {'column_family': 'Standard1'} - while current < end: - start = fmt % current - finish = fmt % last - keyrange = {'start_key': start, 'end_key': finish, 'count': options.rangecount} - begin = time.time() - try: - r = self.cclient.request('get_range_slices', {'column_parent': parent, 'predicate': p, 'range': keyrange, 'consistency_level': consistency}) - if not r: raise RuntimeError("Range not found:", start, finish) - except KeyboardInterrupt: - raise - except Exception, e: - if options.ignore: - print e - else: - print start, finish - raise - current += len(r) + 1 - last = current + len(r) + 1 - self.latencies[self.idx] += time.time() - begin - self.opcounts[self.idx] += 1 - self.keycounts[self.idx] += len(r) - - -class OperationFactory: - @staticmethod - def create(type, i, opcounts, keycounts, latencies): - if type == 'read': - return Reader(i, opcounts, keycounts, latencies) - elif type == 'insert': - return Inserter(i, opcounts, keycounts, latencies) - elif type == 'rangeslice': - return RangeSlicer(i, opcounts, keycounts, latencies) - else: - raise RuntimeError, 'Unsupported op!' - - -class Stress(object): - opcounts = array('i', [0] * n_threads) - latencies = array('d', [0] * n_threads) - keycounts = array('i', [0] * n_threads) - - def create_threads(self,type): - threads = [] - for i in xrange(n_threads): - th = OperationFactory.create(type, i, self.opcounts, self.keycounts, self.latencies) - threads.append(th) - th.start() - return threads - - def run_test(self,filename,threads): - start_t = time.time() - if filename: - outf = open(filename,'w') - else: - outf = sys.stdout - outf.write('total,interval_op_rate,interval_key_rate,avg_latency,elapsed_time\n') - epoch = total = old_total = latency = keycount = old_keycount = old_latency = 0 - epoch_intervals = (options.interval * 10) # 1 epoch = 1 tenth of a second - terminate = False - while not terminate: - time.sleep(0.1) - if not [th for th in threads if th.isAlive()]: - terminate = True - epoch = epoch + 1 - if terminate or epoch > epoch_intervals: - epoch = 0 - old_total, old_latency, old_keycount = total, latency, keycount - total = sum(self.opcounts[th.idx] for th in threads) - latency = sum(self.latencies[th.idx] for th in threads) - keycount = sum(self.keycounts[th.idx] for th in threads) - opdelta = total - old_total - keydelta = keycount - old_keycount - delta_latency = latency - old_latency - if opdelta > 0: - delta_formatted = (delta_latency / opdelta) - else: - delta_formatted = 'NaN' - elapsed_t = int(time.time() - start_t) - outf.write('%d,%d,%d,%s,%d\n' - % (total, opdelta / options.interval, keydelta / options.interval, delta_formatted, elapsed_t)) - - def insert(self): - threads = self.create_threads('insert') - self.run_test(options.file,threads); - - def read(self): - threads = self.create_threads('read') - self.run_test(options.file,threads); - - def rangeslice(self): - threads = self.create_threads('rangeslice') - self.run_test(options.file,threads); - -stresser = Stress() -benchmark = getattr(stresser, options.operation, None) -if not have_multiproc: - print """WARNING: multiprocessing not present, threading will be used. - Benchmark may not be accurate!""" -if options.operation == 'insert': - make_keyspaces() -benchmark() diff --git a/src/java/org/apache/cassandra/avro/AvroErrorFactory.java b/src/java/org/apache/cassandra/avro/AvroErrorFactory.java deleted file mode 100644 index 29fbaed931..0000000000 --- a/src/java/org/apache/cassandra/avro/AvroErrorFactory.java +++ /dev/null @@ -1,94 +0,0 @@ -package org.apache.cassandra.avro; - -import java.util.List; - -import org.apache.avro.util.Utf8; - -public class AvroErrorFactory -{ - public static InvalidRequestException newInvalidRequestException(Utf8 why) - { - InvalidRequestException exception = new InvalidRequestException(); - exception.why = why; - return exception; - } - - public static InvalidRequestException newInvalidRequestException(String why) - { - return newInvalidRequestException(new Utf8(why)); - } - - public static InvalidRequestException newInvalidRequestException(Throwable e) - { - InvalidRequestException exception = newInvalidRequestException(e.getMessage()); - exception.initCause(e); - return exception; - } - - public static NotFoundException newNotFoundException(Utf8 why) - { - NotFoundException exception = new NotFoundException(); - exception.why = why; - return exception; - } - - public static NotFoundException newNotFoundException(String why) - { - return newNotFoundException(new Utf8(why)); - } - - public static NotFoundException newNotFoundException() - { - return newNotFoundException(new Utf8()); - } - - public static TimedOutException newTimedOutException(Utf8 why) - { - TimedOutException exception = new TimedOutException(); - exception.why = why; - return exception; - } - - public static TimedOutException newTimedOutException(String why) - { - return newTimedOutException(new Utf8(why)); - } - - public static TimedOutException newTimedOutException() - { - return newTimedOutException(new Utf8()); - } - - public static UnavailableException newUnavailableException(Utf8 why) - { - UnavailableException exception = new UnavailableException(); - exception.why = why; - return exception; - } - - public static UnavailableException newUnavailableException(String why) - { - return newUnavailableException(new Utf8(why)); - } - - public static UnavailableException newUnavailableException(Throwable t) - { - UnavailableException exception = newUnavailableException(t.getMessage()); - exception.initCause(t); - return exception; - } - - public static UnavailableException newUnavailableException() - { - return newUnavailableException(new Utf8()); - } - - public static TokenRange newTokenRange(String startRange, String endRange, List endpoints) - { - TokenRange tRange = new TokenRange(); - tRange.start_token = startRange; - tRange.end_token = endRange; - tRange.endpoints = (List) endpoints; - return tRange; - } -} diff --git a/src/java/org/apache/cassandra/avro/AvroRecordFactory.java b/src/java/org/apache/cassandra/avro/AvroRecordFactory.java deleted file mode 100644 index dbcecc102a..0000000000 --- a/src/java/org/apache/cassandra/avro/AvroRecordFactory.java +++ /dev/null @@ -1,113 +0,0 @@ -package org.apache.cassandra.avro; -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ - - -import java.nio.ByteBuffer; -import java.util.List; - -import org.apache.avro.generic.GenericArray; -import org.apache.avro.util.Utf8; - -public class AvroRecordFactory -{ - public static Column newColumn(ByteBuffer name, ByteBuffer value, long timestamp) - { - Column column = new Column(); - column.name = name; - column.value = value; - column.timestamp = timestamp; - return column; - } - - public static Column newColumn(byte[] name, byte[] value, long timestamp) - { - return newColumn(ByteBuffer.wrap(name), ByteBuffer.wrap(value), timestamp); - } - - public static SuperColumn newSuperColumn(ByteBuffer name, List columns) - { - SuperColumn column = new SuperColumn(); - column.name = name; - column.columns = columns; - return column; - } - - public static SuperColumn newSuperColumn(byte[] name, List columns) - { - return newSuperColumn(ByteBuffer.wrap(name), columns); - } - - public static ColumnOrSuperColumn newColumnOrSuperColumn(Column column) - { - ColumnOrSuperColumn col = new ColumnOrSuperColumn(); - col.column = column; - return col; - } - - public static ColumnOrSuperColumn newColumnOrSuperColumn(SuperColumn superColumn) - { - ColumnOrSuperColumn column = new ColumnOrSuperColumn(); - column.super_column = superColumn; - return column; - } - - public static ColumnPath newColumnPath(String cfName, ByteBuffer superColumn, ByteBuffer column) - { - ColumnPath cPath = new ColumnPath(); - cPath.column_family = new Utf8(cfName); - cPath.super_column = superColumn; - cPath.column = column; - return cPath; - } - - public static ColumnPath newColumnPath(String cfName, byte[] superColumn, byte[] column) - { - ByteBuffer wrappedSuperColumn = (superColumn != null) ? ByteBuffer.wrap(superColumn) : null; - ByteBuffer wrappedColumn = (column != null) ? ByteBuffer.wrap(column) : null; - return newColumnPath(cfName, wrappedSuperColumn, wrappedColumn); - } - - public static ColumnParent newColumnParent(String cfName, byte[] superColumn) - { - ColumnParent cp = new ColumnParent(); - cp.column_family = new Utf8(cfName); - if (superColumn != null) - cp.super_column = ByteBuffer.wrap(superColumn); - return cp; - } - - public static CoscsMapEntry newCoscsMapEntry(ByteBuffer key, GenericArray columns) - { - CoscsMapEntry entry = new CoscsMapEntry(); - entry.key = key; - entry.columns = columns; - return entry; - } - - public static KeySlice newKeySlice(ByteBuffer key, List columns) { - KeySlice slice = new KeySlice(); - slice.key = key; - slice.columns = columns; - return slice; - } - -} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/avro/AvroValidation.java b/src/java/org/apache/cassandra/avro/AvroValidation.java deleted file mode 100644 index ffd439155c..0000000000 --- a/src/java/org/apache/cassandra/avro/AvroValidation.java +++ /dev/null @@ -1,336 +0,0 @@ -package org.apache.cassandra.avro; -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ - - -import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.Comparator; -import java.util.Set; - -import org.apache.avro.util.Utf8; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.db.ColumnFamily; -import org.apache.cassandra.db.ColumnFamilyType; -import org.apache.cassandra.db.IColumn; -import org.apache.cassandra.db.Table; -import org.apache.cassandra.db.marshal.AbstractType; -import org.apache.cassandra.db.marshal.MarshalException; -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.RandomPartitioner; -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.utils.FBUtilities; - -import static org.apache.cassandra.avro.AvroErrorFactory.newInvalidRequestException; -import static org.apache.cassandra.avro.AvroRecordFactory.newColumnPath; - -/** - * The Avro analogue to org.apache.cassandra.service.ThriftValidation - */ -public class AvroValidation -{ - public static void validateKey(ByteBuffer key) throws InvalidRequestException - { - if (key == null || key.remaining() == 0) - throw newInvalidRequestException("Key may not be empty"); - - // check that key can be handled by FBUtilities.writeShortByteArray - if (key.remaining() > FBUtilities.MAX_UNSIGNED_SHORT) - throw newInvalidRequestException("Key length of " + key.remaining() + - " is longer than maximum of " + FBUtilities.MAX_UNSIGNED_SHORT); - } - - - // FIXME: could use method in ThriftValidation - static void validateKeyspace(String keyspace) throws KeyspaceNotDefinedException - { - if (!DatabaseDescriptor.getTables().contains(keyspace)) - throw new KeyspaceNotDefinedException(new Utf8("Keyspace " + keyspace + " does not exist in this schema.")); - } - - // FIXME: could use method in ThriftValidation - public static ColumnFamilyType validateColumnFamily(String keyspace, String columnFamily) throws InvalidRequestException - { - if (columnFamily.isEmpty()) - throw newInvalidRequestException("non-empty columnfamily is required"); - - ColumnFamilyType cfType = DatabaseDescriptor.getColumnFamilyType(keyspace, columnFamily); - if (cfType == null) - throw newInvalidRequestException("unconfigured columnfamily " + columnFamily); - - return cfType; - } - - static void validateColumnPath(String keyspace, ColumnPath cp) throws InvalidRequestException - { - validateKeyspace(keyspace); - String column_family = cp.column_family.toString(); - ColumnFamilyType cfType = validateColumnFamily(keyspace, column_family); - - - if (cfType == ColumnFamilyType.Standard) - { - if (cp.super_column != null) - throw newInvalidRequestException("supercolumn parameter is invalid for standard CF " + column_family); - - if (cp.column == null) - throw newInvalidRequestException("column parameter is not optional for standard CF " + column_family); - } - else - { - if (cp.super_column == null) - throw newInvalidRequestException("supercolumn parameter is not optional for super CF " + column_family); - } - - if (cp.column != null) - validateColumns(keyspace, column_family, cp.super_column, Arrays.asList(cp.column)); - if (cp.super_column != null) - validateColumns(keyspace, column_family, null, Arrays.asList(cp.super_column)); - } - - static void validateColumnParent(String keyspace, ColumnParent parent) throws InvalidRequestException - { - validateKeyspace(keyspace); - String cfName = parent.column_family.toString(); - ColumnFamilyType cfType = validateColumnFamily(keyspace, cfName); - - if (cfType == ColumnFamilyType.Standard) - if (parent.super_column != null) - throw newInvalidRequestException("super column specified for standard column family"); - if (parent.super_column != null) - validateColumns(keyspace, cfName, null, Arrays.asList(parent.super_column)); - } - - // FIXME: could use method in ThriftValidation - static void validateColumns(String keyspace, String cfName, ByteBuffer superColumnName, Iterable columnNames) - throws InvalidRequestException - { - if (superColumnName != null) - { - if (superColumnName.remaining() > IColumn.MAX_NAME_LENGTH) - throw newInvalidRequestException("supercolumn name length must not be greater than " + IColumn.MAX_NAME_LENGTH); - if (superColumnName.remaining() == 0) - throw newInvalidRequestException("supercolumn name must not be empty"); - if (DatabaseDescriptor.getColumnFamilyType(keyspace, cfName) == ColumnFamilyType.Standard) - throw newInvalidRequestException("supercolumn specified to ColumnFamily " + cfName + " containing normal columns"); - } - - AbstractType comparator = ColumnFamily.getComparatorFor(keyspace, cfName, superColumnName); - for (ByteBuffer buff : columnNames) - { - - if (buff.remaining() > IColumn.MAX_NAME_LENGTH) - throw newInvalidRequestException("column name length must not be greater than " + IColumn.MAX_NAME_LENGTH); - if (buff.remaining() == 0) - throw newInvalidRequestException("column name must not be empty"); - - try - { - comparator.validate(buff); - } - catch (MarshalException e) - { - throw newInvalidRequestException(e.getMessage()); - } - } - } - - static void validateColumns(String keyspace, ColumnParent parent, Iterable columnNames) - throws InvalidRequestException - { - validateColumns(keyspace, - parent.column_family.toString(), - parent.super_column, - columnNames); - } - - static void validateColumn(String keyspace, ColumnParent parent, Column column) - throws InvalidRequestException - { - validateTtl(column); - validateColumns(keyspace, parent, Arrays.asList(column.name)); - } - - static void validateColumnOrSuperColumn(String keyspace, String cfName, ColumnOrSuperColumn cosc) - throws InvalidRequestException - { - if (cosc.column != null) - AvroValidation.validateColumnPath(keyspace, newColumnPath(cfName, null, cosc.column.name)); - - if (cosc.super_column != null) - for (Column c : cosc.super_column.columns) - AvroValidation.validateColumnPath(keyspace, newColumnPath(cfName, cosc.super_column.name, c.name)); - - if ((cosc.column == null) && (cosc.super_column == null)) - throw newInvalidRequestException("ColumnOrSuperColumn must have one or both of Column or SuperColumn"); - } - - static void validateRange(String keyspace, String cfName, ByteBuffer superName, SliceRange range) - throws InvalidRequestException - { - AbstractType comparator = ColumnFamily.getComparatorFor(keyspace, cfName, superName); - - - try - { - comparator.validate(range.start); - comparator.validate(range.finish); - } - catch (MarshalException me) - { - throw newInvalidRequestException(me.getMessage()); - } - - if (range.count < 0) - throw newInvalidRequestException("Ranges require a non-negative count."); - - Comparator orderedComparator = range.reversed ? comparator.getReverseComparator() : comparator; - if (range.start.remaining() > 0 && range.finish.remaining() > 0 && orderedComparator.compare(range.start, range.finish) > 0) - throw newInvalidRequestException("range finish must come after start in the order of traversal"); - } - - static void validateRange(String keyspace, ColumnParent cp, SliceRange range) throws InvalidRequestException - { - validateRange(keyspace, cp.column_family.toString(), cp.super_column, range); - } - - static void validateSlicePredicate(String keyspace, String cfName, ByteBuffer superName, SlicePredicate predicate) - throws InvalidRequestException - { - if (predicate.column_names == null && predicate.slice_range == null) - throw newInvalidRequestException("A SlicePredicate must be given a list of Columns, a SliceRange, or both"); - - if (predicate.slice_range != null) - validateRange(keyspace, cfName, superName, predicate.slice_range); - - if (predicate.column_names != null) - validateColumns(keyspace, cfName, superName, predicate.column_names); - } - - static void validateDeletion(String keyspace, String cfName, Deletion del) throws InvalidRequestException - { - validateColumnFamily(keyspace, cfName); - if (del.super_column == null && del.predicate == null) - throw newInvalidRequestException("A Deletion must have a SuperColumn, a SlicePredicate, or both."); - - if (del.predicate != null) - { - validateSlicePredicate(keyspace, cfName, del.super_column, del.predicate); - if (del.predicate.slice_range != null) - throw newInvalidRequestException("Deletion does not yet support SliceRange predicates."); - } - } - - static void validateMutation(String keyspace, String cfName, Mutation mutation) throws InvalidRequestException - { - ColumnOrSuperColumn cosc = mutation.column_or_supercolumn; - Deletion del = mutation.deletion; - - if (cosc != null && del != null) - throw newInvalidRequestException("Mutation may have either a ColumnOrSuperColumn or a Deletion, but not both"); - - if (cosc != null) - { - validateColumnOrSuperColumn(keyspace, cfName, cosc); - } - else if (del != null) - { - validateDeletion(keyspace, cfName, del); - } - else - { - throw newInvalidRequestException("Mutation must have one ColumnOrSuperColumn, or one Deletion"); - } - } - - static void validateTtl(Column column) throws InvalidRequestException - { - if (column.ttl != null && column.ttl < 0) - throw newInvalidRequestException("ttl must be a positive value"); - } - - static void validatePredicate(String keyspace, ColumnParent cp, SlicePredicate predicate) - throws InvalidRequestException - { - if (predicate.column_names == null && predicate.slice_range == null) - throw newInvalidRequestException("predicate column_names and slice_range may not both be null"); - - if (predicate.column_names != null && predicate.slice_range != null) - throw newInvalidRequestException("predicate column_names and slice_range may not both be set"); - - if (predicate.slice_range != null) - validateRange(keyspace, cp, predicate.slice_range); - else - validateColumns(keyspace, cp, predicate.column_names); - } - - public static void validateKeyRange(KeyRange range) - throws InvalidRequestException - { - if ((range.start_key == null) != (range.end_key == null)) - { - throw newInvalidRequestException("start key and end key must either both be non-null, or both be null"); - } - if ((range.start_token == null) != (range.end_token == null)) - { - throw newInvalidRequestException("start token and end token must either both be non-null, or both be null"); - } - if ((range.start_key == null) == (range.start_token == null)) - { - throw newInvalidRequestException("exactly one of {start key, end key} or {start token, end token} must be specified"); - } - - if (range.start_key != null) - { - IPartitioner p = StorageService.getPartitioner(); - Token startToken = p.getToken(range.start_key); - Token endToken = p.getToken(range.end_key); - if (startToken.compareTo(endToken) > 0 && !endToken.equals(p.getMinimumToken())) - { - if (p instanceof RandomPartitioner) - throw newInvalidRequestException("start key's md5 sorts after end key's md5. this is not allowed; you probably should not specify end key at all, under RandomPartitioner"); - else - throw newInvalidRequestException("start key must sort before (or equal to) finish key in your partitioner!"); - } - } - - if (range.count <= 0) - { - throw newInvalidRequestException("maxRows must be positive"); - } - } - - static void validateIndexClauses(String keyspace, String columnFamily, IndexClause index_clause) - throws InvalidRequestException - { - if (index_clause.expressions.isEmpty()) - throw newInvalidRequestException("index clause list may not be empty"); - Set indexedColumns = Table.open(keyspace).getColumnFamilyStore(columnFamily).getIndexedColumns(); - for (IndexExpression expression : index_clause.expressions) - { - if (expression.op.equals(IndexOperator.EQ) && indexedColumns.contains(expression.column_name)) - return; - } - throw newInvalidRequestException("No indexed columns present in index clause with operator EQ"); - } - -} diff --git a/src/java/org/apache/cassandra/avro/CassandraDaemon.java b/src/java/org/apache/cassandra/avro/CassandraDaemon.java deleted file mode 100644 index 460e1b05a0..0000000000 --- a/src/java/org/apache/cassandra/avro/CassandraDaemon.java +++ /dev/null @@ -1,85 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.cassandra.avro; - -import java.io.IOException; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.avro.ipc.ResponderServlet; -import org.apache.avro.specific.SpecificResponder; -import org.mortbay.jetty.servlet.Context; -import org.mortbay.jetty.servlet.ServletHolder; - -/** - * The Avro analogue to org.apache.cassandra.service.CassandraDaemon. - * - */ -public class CassandraDaemon extends org.apache.cassandra.service.AbstractCassandraDaemon { - private static Logger logger = LoggerFactory.getLogger(CassandraDaemon.class); - private org.mortbay.jetty.Server server; - - /** hook for JSVC */ - public void start() throws IOException - { - if (logger.isDebugEnabled()) - logger.debug(String.format("Binding avro service to %s:%s", listenAddr, listenPort)); - CassandraServer cassandraServer = new CassandraServer(); - SpecificResponder responder = new SpecificResponder(Cassandra.class, cassandraServer); - - logger.info("Listening for avro clients..."); - - // FIXME: This isn't actually binding to listenAddr (it should). - server = new org.mortbay.jetty.Server(listenPort); - server.setThreadPool(new CleaningThreadPool(cassandraServer.clientState, - MIN_WORKER_THREADS, - Integer.MAX_VALUE)); - try - { - // see CASSANDRA-1440 - ResponderServlet servlet = new ResponderServlet(responder); - new Context(server, "/").addServlet(new ServletHolder(servlet), "/*"); - - server.start(); - } - catch (Exception e) - { - throw new IOException("Could not start Avro server.", e); - } - } - - /** hook for JSVC */ - public void stop() - { - logger.info("Cassandra shutting down..."); - try - { - server.stop(); - } - catch (Exception e) - { - logger.error("Avro server did not exit cleanly.", e); - } - } - - public static void main(String[] args) { - new CassandraDaemon().activate(); - } -} diff --git a/src/java/org/apache/cassandra/avro/CassandraServer.java b/src/java/org/apache/cassandra/avro/CassandraServer.java deleted file mode 100644 index d68ed8c978..0000000000 --- a/src/java/org/apache/cassandra/avro/CassandraServer.java +++ /dev/null @@ -1,1153 +0,0 @@ -package org.apache.cassandra.avro; -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ - -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.UnsupportedEncodingException; -import java.nio.ByteBuffer; -import java.util.*; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.TimeoutException; -import java.util.zip.DataFormatException; -import java.util.zip.Inflater; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.antlr.runtime.RecognitionException; -import org.apache.avro.Schema; -import org.apache.avro.generic.GenericArray; -import org.apache.avro.generic.GenericData; -import org.apache.avro.ipc.AvroRemoteException; -import org.apache.avro.util.Utf8; -import org.apache.cassandra.auth.AllowAllAuthenticator; -import org.apache.cassandra.auth.Permission; -import org.apache.cassandra.concurrent.Stage; -import org.apache.cassandra.concurrent.StageManager; -import org.apache.cassandra.config.*; -import org.apache.cassandra.cql.QueryProcessor; -import org.apache.cassandra.db.*; -import org.apache.cassandra.db.filter.QueryPath; -import org.apache.cassandra.db.marshal.MarshalException; -import org.apache.cassandra.db.migration.*; -import org.apache.cassandra.dht.*; -import org.apache.cassandra.locator.AbstractReplicationStrategy; -import org.apache.cassandra.scheduler.IRequestScheduler; -import org.apache.cassandra.service.ClientState; -import org.apache.cassandra.service.StorageProxy; -import org.apache.cassandra.service.StorageService; - -import static org.apache.cassandra.avro.AvroErrorFactory.*; -import static org.apache.cassandra.avro.AvroRecordFactory.*; - -public class CassandraServer implements Cassandra { - private static Logger logger = LoggerFactory.getLogger(CassandraServer.class); - - private final static GenericArray EMPTY_SUBCOLUMNS = new GenericData.Array(0, Schema.createArray(Column.SCHEMA$)); - private final static GenericArray EMPTY_COLUMNS = new GenericData.Array(0, Schema.createArray(ColumnOrSuperColumn.SCHEMA$)); - private final static Utf8 API_VERSION = new Utf8("0.0.0"); - - // CfDef default values - private final static String D_CF_CFTYPE = "Standard"; - private final static String D_CF_COMPTYPE = "BytesType"; - private final static String D_CF_SUBCOMPTYPE = ""; - private final static String D_CF_RECONCILER = null; - - //ColumnDef default values - public final static String D_COLDEF_INDEXTYPE = "KEYS"; - public final static String D_COLDEF_INDEXNAME = null; - - // thread local state containing session information - public final ThreadLocal clientState = new ThreadLocal() - { - @Override - public ClientState initialValue() - { - return new ClientState(); - } - }; - - /* - * RequestScheduler to perform the scheduling of incoming requests - */ - private final IRequestScheduler requestScheduler; - - public CassandraServer() - { - requestScheduler = DatabaseDescriptor.getRequestScheduler(); - } - - public Void login(AuthenticationRequest auth_request) throws AuthenticationException, AuthorizationException - { - try - { - state().login(auth_request.credentials); - } - catch (org.apache.cassandra.thrift.AuthenticationException thriftE) - { - throw new AuthenticationException(); - } - return null; - } - - public ClientState state() - { - return clientState.get(); - } - - @Override - public ColumnOrSuperColumn get(ByteBuffer key, ColumnPath columnPath, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, InvalidRequestException, NotFoundException, UnavailableException, TimedOutException { - if (logger.isDebugEnabled()) - logger.debug("get"); - - AvroValidation.validateColumnPath(state().getKeyspace(), columnPath); - - // FIXME: This is repetitive. - ByteBuffer column, super_column; - column = columnPath.column == null ? null : columnPath.column; - super_column = columnPath.super_column == null ? null : columnPath.super_column; - - QueryPath path = new QueryPath(columnPath.column_family.toString(), column == null ? null : super_column); - List nameAsList = Arrays.asList(column == null ? super_column : column); - AvroValidation.validateKey(key); - ReadCommand command = new SliceByNamesReadCommand(state().getKeyspace(), key, path, nameAsList); - - Map, ColumnFamily> cfamilies = readColumnFamily(Arrays.asList(command), consistencyLevel); - ColumnFamily cf = cfamilies.get(StorageService.getPartitioner().decorateKey(command.key)); - - if (cf == null) - throw newNotFoundException(); - - GenericArray avroColumns = avronateColumnFamily(cf, command.queryPath.superColumnName != null, false); - - if (avroColumns.size() == 0) - throw newNotFoundException(); - - assert avroColumns.size() == 1; - return avroColumns.iterator().next(); - } - - protected Map, ColumnFamily> readColumnFamily(List commands, ConsistencyLevel consistency) - throws InvalidRequestException, UnavailableException, TimedOutException - { - // TODO - Support multiple column families per row, right now row only contains 1 column family - Map, ColumnFamily> columnFamilyKeyMap = new HashMap, ColumnFamily>(); - - List rows; - try - { - schedule(); - rows = StorageProxy.readProtocol(commands, thriftConsistencyLevel(consistency)); - } - catch (TimeoutException e) - { - throw new TimedOutException(); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - // FIXME: This suckage brought to you by StorageService and StorageProxy - // which throw Thrift exceptions directly. - catch (org.apache.cassandra.thrift.UnavailableException e) - { - throw newUnavailableException(e); - } - catch (org.apache.cassandra.thrift.InvalidRequestException e) - { - throw newInvalidRequestException(e); - } - finally - { - release(); - } - - for (Row row: rows) - { - columnFamilyKeyMap.put(row.key, row.cf); - } - - return columnFamilyKeyMap; - } - - // Don't playa hate, avronate. - private List avronateSubColumns(Collection columns) - { - if (columns == null || columns.isEmpty()) - return EMPTY_SUBCOLUMNS; - - List avroColumns = new ArrayList(columns.size()); - - for (IColumn column : columns) - { - if (column.isMarkedForDelete()) - continue; - - Column avroColumn = newColumn(column.name(), column.value(), column.timestamp()); - avroColumns.add(avroColumn); - } - - return avroColumns; - } - - private GenericArray avronateColumns(Collection columns, boolean reverseOrder) - { - ArrayList avroColumns = new ArrayList(columns.size()); - for (IColumn column : columns) - { - if (column.isMarkedForDelete()) - continue; - - Column avroColumn = newColumn(column.name(), column.value(), column.timestamp()); - - if (column instanceof ExpiringColumn) - avroColumn.ttl = ((ExpiringColumn)column).getTimeToLive(); - - avroColumns.add(newColumnOrSuperColumn(avroColumn)); - } - - if (reverseOrder) - Collections.reverse(avroColumns); - - // FIXME: update for AVRO-540 when upgrading to Avro 1.4.0 - GenericArray avroArray = new GenericData.Array(avroColumns.size(), Schema.createArray(ColumnOrSuperColumn.SCHEMA$)); - for (ColumnOrSuperColumn cosc : avroColumns) - avroArray.add(cosc); - - return avroArray; - } - - private GenericArray avronateSuperColumns(Collection columns, boolean reverseOrder) - { - ArrayList avroSuperColumns = new ArrayList(columns.size()); - for (IColumn column: columns) - { - List subColumns = avronateSubColumns(column.getSubColumns()); - if (subColumns.size() == 0) - continue; - SuperColumn superColumn = newSuperColumn(column.name(), subColumns); - avroSuperColumns.add(newColumnOrSuperColumn(superColumn)); - } - - if (reverseOrder) - Collections.reverse(avroSuperColumns); - - // FIXME: update for AVRO-540 when upgrading to Avro 1.4.0 - GenericArray avroArray = new GenericData.Array(avroSuperColumns.size(), Schema.createArray(ColumnOrSuperColumn.SCHEMA$)); - for (ColumnOrSuperColumn cosc : avroSuperColumns) - avroArray.add(cosc); - - return avroArray; - } - - private GenericArray avronateColumnFamily(ColumnFamily cf, boolean subColumnsOnly, boolean reverseOrder) - { - if (cf == null || cf.getColumnsMap().size() == 0) - return EMPTY_COLUMNS; - - if (subColumnsOnly) - { - IColumn column = cf.getColumnsMap().values().iterator().next(); - Collection subColumns = column.getSubColumns(); - if (subColumns == null || subColumns.isEmpty()) - return EMPTY_COLUMNS; - else - return avronateColumns(subColumns, reverseOrder); - } - - if (cf.isSuper()) - return avronateSuperColumns(cf.getSortedColumns(), reverseOrder); - else - return avronateColumns(cf.getSortedColumns(), reverseOrder); - } - - public List get_slice(ByteBuffer key, ColumnParent columnParent, - SlicePredicate predicate, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, InvalidRequestException, UnavailableException, TimedOutException - { - if (logger.isDebugEnabled()) - logger.debug("get_slice"); - - Schema bytesArray = Schema.createArray(Schema.parse("{\"type\": \"bytes\"}")); - GenericArray keys = new GenericData.Array(1, bytesArray); - keys.add(key); - - return multigetSliceInternal(state().getKeyspace(), keys, columnParent, predicate, consistencyLevel).iterator().next().columns; - } - - private List multigetSliceInternal(String keyspace, List keys, - ColumnParent columnParent, SlicePredicate predicate, ConsistencyLevel consistencyLevel) - throws InvalidRequestException, UnavailableException, TimedOutException - { - AvroValidation.validateColumnParent(keyspace, columnParent); - AvroValidation.validatePredicate(keyspace, columnParent, predicate); - - QueryPath queryPath = new QueryPath(columnParent.column_family.toString(), columnParent.super_column); - - List commands = new ArrayList(); - if (predicate.column_names != null) - { - for (ByteBuffer key : keys) - { - AvroValidation.validateKey(key); - - commands.add(new SliceByNamesReadCommand(keyspace, key, queryPath, predicate.column_names)); - } - } - else - { - SliceRange range = predicate.slice_range; - for (ByteBuffer key : keys) - { - AvroValidation.validateKey(key); - commands.add(new SliceFromReadCommand(keyspace, key, queryPath, range.start, range.finish, range.reversed, range.count)); - } - } - - return getSlice(commands, consistencyLevel); - } - - private List getSlice(List commands, ConsistencyLevel consistencyLevel) - throws InvalidRequestException, UnavailableException, TimedOutException - { - Map, ColumnFamily> columnFamilies = readColumnFamily(commands, consistencyLevel); - Schema sch = Schema.createArray(CoscsMapEntry.SCHEMA$); - List columnFamiliesList = new GenericData.Array(commands.size(), sch); - - for (ReadCommand cmd : commands) - { - ColumnFamily cf = columnFamilies.get(StorageService.getPartitioner().decorateKey(cmd.key)); - boolean reverseOrder = cmd instanceof SliceFromReadCommand && ((SliceFromReadCommand)cmd).reversed; - GenericArray avroColumns = avronateColumnFamily(cf, cmd.queryPath.superColumnName != null, reverseOrder); - columnFamiliesList.add(newCoscsMapEntry(cmd.key, avroColumns)); - } - - return columnFamiliesList; - } - - public int get_count(ByteBuffer key, ColumnParent columnParent, SlicePredicate predicate, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, InvalidRequestException, UnavailableException, TimedOutException - { - if (logger.isDebugEnabled()) - logger.debug("get_count"); - - return (int)get_slice(key, columnParent, predicate, consistencyLevel).size(); - } - - public List multiget_slice(List keys, ColumnParent columnParent, - SlicePredicate predicate, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, InvalidRequestException, UnavailableException, TimedOutException - { - if (logger.isDebugEnabled()) - logger.debug("multiget_slice"); - - return multigetSliceInternal(state().getKeyspace(), keys, columnParent, predicate, consistencyLevel); - } - - public Void insert(ByteBuffer key, ColumnParent parent, Column column, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, InvalidRequestException, UnavailableException, TimedOutException - { - if (logger.isDebugEnabled()) - logger.debug("insert"); - - AvroValidation.validateKey(key); - AvroValidation.validateColumnParent(state().getKeyspace(), parent); - AvroValidation.validateColumn(state().getKeyspace(), parent, column); - - RowMutation rm = new RowMutation(state().getKeyspace(), key); - try - { - rm.add(new QueryPath(parent.column_family.toString(), - parent.super_column, - column.name), - column.value, - column.timestamp, - column.ttl == null ? 0 : column.ttl); - } - catch (MarshalException e) - { - throw newInvalidRequestException(e.getMessage()); - } - doInsert(consistencyLevel, rm); - - return null; - } - - public Void remove(ByteBuffer key, ColumnPath columnPath, long timestamp, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, InvalidRequestException, UnavailableException, TimedOutException - { - if (logger.isDebugEnabled()) - logger.debug("remove"); - - AvroValidation.validateKey(key); - AvroValidation.validateColumnPath(state().getKeyspace(), columnPath); - - RowMutation rm = new RowMutation(state().getKeyspace(), key); - rm.delete(new QueryPath(columnPath.column_family.toString(), columnPath.super_column), timestamp); - - doInsert(consistencyLevel, rm); - - return null; - } - - private void doInsert(ConsistencyLevel consistency, RowMutation rm) throws UnavailableException, TimedOutException - { - try - { - schedule(); - StorageProxy.mutate(Arrays.asList(rm), thriftConsistencyLevel(consistency)); - } - catch (TimeoutException e) - { - throw new TimedOutException(); - } - catch (org.apache.cassandra.thrift.UnavailableException thriftE) - { - throw newUnavailableException(thriftE); - } - finally - { - release(); - } - } - - public Void batch_mutate(List mutationMap, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, InvalidRequestException, UnavailableException, TimedOutException - { - if (logger.isDebugEnabled()) - logger.debug("batch_mutate"); - - List rowMutations = new ArrayList(); - - for (MutationsMapEntry pair: mutationMap) - { - AvroValidation.validateKey(pair.key); - Map> cfToMutations = pair.mutations; - - for (Map.Entry> cfMutations : cfToMutations.entrySet()) - { - String cfName = cfMutations.getKey().toString(); - - for (Mutation mutation : cfMutations.getValue()) - AvroValidation.validateMutation(state().getKeyspace(), cfName, mutation); - } - rowMutations.add(getRowMutationFromMutations(state().getKeyspace(), pair.key, cfToMutations)); - } - - try - { - schedule(); - StorageProxy.mutate(rowMutations, thriftConsistencyLevel(consistencyLevel)); - } - catch (TimeoutException te) - { - throw newTimedOutException(); - } - // FIXME: StorageProxy.mutate throws Thrift's UnavailableException - catch (org.apache.cassandra.thrift.UnavailableException ue) - { - throw newUnavailableException(); - } - finally - { - release(); - } - - return null; - } - - // FIXME: This is copypasta from o.a.c.db.RowMutation, (RowMutation.getRowMutation uses Thrift types directly). - private static RowMutation getRowMutationFromMutations(String keyspace, ByteBuffer key, Map> cfMap) - { - RowMutation rm = new RowMutation(keyspace, key); - - for (Map.Entry> entry : cfMap.entrySet()) - { - String cfName = entry.getKey().toString(); - - for (Mutation mutation : entry.getValue()) - { - if (mutation.deletion != null) - deleteColumnOrSuperColumnToRowMutation(rm, cfName, mutation.deletion); - else - addColumnOrSuperColumnToRowMutation(rm, cfName, mutation.column_or_supercolumn); - } - } - - return rm; - } - - // FIXME: This is copypasta from o.a.c.db.RowMutation, (RowMutation.getRowMutation uses Thrift types directly). - private static void addColumnOrSuperColumnToRowMutation(RowMutation rm, String cfName, ColumnOrSuperColumn cosc) - { - if (cosc.column == null) - { - for (Column column : cosc.super_column.columns) - rm.add(new QueryPath(cfName, cosc.super_column.name, column.name), column.value, column.timestamp); - } - else - { - rm.add(new QueryPath(cfName, null, cosc.column.name), cosc.column.value, cosc.column.timestamp); - } - } - - // 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) - { - - if (del.predicate != null && del.predicate.column_names != null) - { - for (ByteBuffer col : del.predicate.column_names) - { - if (del.super_column == null && DatabaseDescriptor.getColumnFamilyType(rm.getTable(), cfName) == ColumnFamilyType.Super) - rm.delete(new QueryPath(cfName, col), del.timestamp); - else - rm.delete(new QueryPath(cfName, del.super_column, col), del.timestamp); - } - } - else - { - rm.delete(new QueryPath(cfName, del.super_column), del.timestamp); - } - } - - // Copy-pasted from the thrift CassandraServer, using the factory methods to create exceptions. - // helper method to apply migration on the migration stage. typical migration failures will throw an - // InvalidRequestException. atypical failures will throw a RuntimeException. - private static void applyMigrationOnStage(final Migration m) throws InvalidRequestException - { - Future f = StageManager.getStage(Stage.MIGRATION).submit(new Callable() - { - public Object call() throws Exception - { - m.apply(); - m.announce(); - return null; - } - }); - try - { - f.get(); - } - catch (InterruptedException e) - { - throw new RuntimeException(e); - } - catch (ExecutionException e) - { - // this means call() threw an exception. deal with it directly. - if (e.getCause() != null) - { - throw newInvalidRequestException(e.getCause()); - } - else - { - throw newInvalidRequestException(e); - } - } - } - - private org.apache.cassandra.thrift.ConsistencyLevel thriftConsistencyLevel(ConsistencyLevel consistency) - { - switch (consistency) - { - case ONE: return org.apache.cassandra.thrift.ConsistencyLevel.ONE; - case QUORUM: return org.apache.cassandra.thrift.ConsistencyLevel.QUORUM; - case LOCAL_QUORUM: return org.apache.cassandra.thrift.ConsistencyLevel.LOCAL_QUORUM; - case EACH_QUORUM: return org.apache.cassandra.thrift.ConsistencyLevel.EACH_QUORUM; - case ALL: return org.apache.cassandra.thrift.ConsistencyLevel.ALL; - } - return null; - } - - public Void set_keyspace(CharSequence keyspace) throws InvalidRequestException - { - String keyspaceStr = keyspace.toString(); - - if (DatabaseDescriptor.getTableDefinition(keyspaceStr) == null) - { - throw newInvalidRequestException("Keyspace does not exist"); - } - - state().setKeyspace(keyspaceStr); - return null; - } - - public CharSequence system_add_keyspace(KsDef ksDef) throws AvroRemoteException, InvalidRequestException - { - if (!(DatabaseDescriptor.getAuthenticator() instanceof AllowAllAuthenticator)) - throw newInvalidRequestException("Unable to create new keyspace while authentication is enabled."); - - //generate a meaningful error if the user setup keyspace and/or column definition incorrectly - for (CfDef cf : ksDef.cf_defs) - { - if (!cf.keyspace.equals(ksDef.name)) - { - throw newInvalidRequestException("CsDef (" + cf.name +") had a keyspace definition that did not match KsDef"); - } - } - - try - { - Collection cfDefs = new ArrayList((int)ksDef.cf_defs.size()); - for (CfDef cfDef : ksDef.cf_defs) - { - cfDefs.add(convertToCFMetaData(cfDef)); - } - - // convert Map to Map - Map strategyOptions = null; - if (ksDef.strategy_options != null && !ksDef.strategy_options.isEmpty()) - { - strategyOptions = new HashMap(); - for (Map.Entry option : ksDef.strategy_options.entrySet()) - { - strategyOptions.put(option.getKey().toString(), option.getValue().toString()); - } - } - - KSMetaData ksmeta = new KSMetaData( - ksDef.name.toString(), - AbstractReplicationStrategy.getClass(ksDef.strategy_class.toString()), - strategyOptions, - ksDef.replication_factor, - cfDefs.toArray(new CFMetaData[cfDefs.size()])); - applyMigrationOnStage(new AddKeyspace(ksmeta)); - return DatabaseDescriptor.getDefsVersion().toString(); - - } - catch (ConfigurationException e) - { - throw newInvalidRequestException(e); - } - catch (IOException e) - { - throw newInvalidRequestException(e); - } - } - - public CharSequence system_add_column_family(CfDef cfDef) throws AvroRemoteException, InvalidRequestException - { - checkKeyspaceAndLoginAuthorized(Permission.WRITE); - try - { - applyMigrationOnStage(new AddColumnFamily(convertToCFMetaData(cfDef))); - return DatabaseDescriptor.getDefsVersion().toString(); - } catch (ConfigurationException e) - { - throw newInvalidRequestException(e); - } - catch (IOException e) - { - throw newInvalidRequestException(e); - } - } - - public CharSequence system_update_column_family(CfDef cf_def) throws AvroRemoteException, InvalidRequestException - { - checkKeyspaceAndLoginAuthorized(Permission.WRITE); - - if (cf_def.keyspace == null || cf_def.name == null) - throw newInvalidRequestException("Keyspace and CF name must be set."); - - CFMetaData oldCfm = DatabaseDescriptor.getCFMetaData(CFMetaData.getId(cf_def.keyspace.toString(), cf_def.name.toString())); - if (oldCfm == null) - throw newInvalidRequestException("Could not find column family definition to modify."); - - try - { - CFMetaData.applyImplicitDefaults(cf_def); - UpdateColumnFamily update = new UpdateColumnFamily(cf_def); - applyMigrationOnStage(update); - return DatabaseDescriptor.getDefsVersion().toString(); - } - catch (ConfigurationException e) - { - InvalidRequestException ex = newInvalidRequestException(e.getMessage()); - ex.initCause(e); - throw ex; - } - catch (IOException e) - { - InvalidRequestException ex = newInvalidRequestException(e.getMessage()); - ex.initCause(e); - throw ex; - } - } - - public CharSequence system_update_keyspace(KsDef ks_def) throws AvroRemoteException, InvalidRequestException - { - checkKeyspaceAndLoginAuthorized(Permission.WRITE); - - if (ks_def.cf_defs != null && ks_def.cf_defs.size() > 0) - throw newInvalidRequestException("Keyspace update must not contain any column family definitions."); - - if (DatabaseDescriptor.getTableDefinition(ks_def.name.toString()) == null) - throw newInvalidRequestException("Keyspace does not exist."); - - try - { - // convert Map to Map - Map strategyOptions = null; - if (ks_def.strategy_options != null && !ks_def.strategy_options.isEmpty()) - { - strategyOptions = new HashMap(); - for (Map.Entry option : ks_def.strategy_options.entrySet()) - { - strategyOptions.put(option.getKey().toString(), option.getValue().toString()); - } - } - - KSMetaData ksm = new KSMetaData( - ks_def.name.toString(), - AbstractReplicationStrategy.getClass(ks_def.strategy_class.toString()), - strategyOptions, - ks_def.replication_factor); - applyMigrationOnStage(new UpdateKeyspace(ksm)); - return DatabaseDescriptor.getDefsVersion().toString(); - } - catch (ConfigurationException e) - { - InvalidRequestException ex = newInvalidRequestException(e.getMessage()); - ex.initCause(e); - throw ex; - } - catch (IOException e) - { - InvalidRequestException ex = newInvalidRequestException(e.getMessage()); - ex.initCause(e); - throw ex; - } - } - - public GenericArray describe_keyspaces() throws AvroRemoteException - { - Set keyspaces = DatabaseDescriptor.getTables(); - Schema schema = Schema.createArray(Schema.create(Schema.Type.STRING)); - GenericArray avroResults = new GenericData.Array(keyspaces.size(), schema); - - for (String ksp : keyspaces) - avroResults.add(new Utf8(ksp)); - - return avroResults; - } - - public Utf8 describe_cluster_name() throws AvroRemoteException - { - return new Utf8(DatabaseDescriptor.getClusterName()); - } - - - public Utf8 describe_version() throws AvroRemoteException - { - return API_VERSION; - } - - public Map> check_schema_agreement() - { - logger.debug("checking schema agreement"); - return (Map) StorageProxy.describeSchemaVersions(); - } - - protected void checkKeyspaceAndLoginAuthorized(Permission perm) throws InvalidRequestException - { - try - { - state().hasColumnFamilyListAccess(perm); - } - catch (org.apache.cassandra.thrift.InvalidRequestException e) - { - throw newInvalidRequestException(e.getWhy()); - } - } - - /** - * Schedule the current thread for access to the required services - */ - private void schedule() - { - requestScheduler.queue(Thread.currentThread(), state().getSchedulingValue()); - } - - /** - * Release a count of resources used to the request scheduler - */ - private void release() - { - requestScheduler.release(); - } - - private CFMetaData convertToCFMetaData(CfDef cf_def) throws InvalidRequestException, ConfigurationException - { - String cfType = cf_def.column_type == null ? D_CF_CFTYPE : cf_def.column_type.toString(); - String compare = cf_def.comparator_type == null ? D_CF_COMPTYPE : cf_def.comparator_type.toString(); - String validate = cf_def.default_validation_class == null ? D_CF_COMPTYPE : cf_def.default_validation_class.toString(); - String subCompare = cf_def.subcomparator_type == null ? D_CF_SUBCOMPTYPE : cf_def.subcomparator_type.toString(); - - CFMetaData.validateMinMaxCompactionThresholds(cf_def); - CFMetaData.validateMemtableSettings(cf_def); - - return new CFMetaData(cf_def.keyspace.toString(), - cf_def.name.toString(), - ColumnFamilyType.create(cfType), - DatabaseDescriptor.getComparator(compare), - subCompare.length() == 0 ? null : DatabaseDescriptor.getComparator(subCompare), - cf_def.comment == null ? "" : cf_def.comment.toString(), - cf_def.row_cache_size == null ? CFMetaData.DEFAULT_ROW_CACHE_SIZE : cf_def.row_cache_size, - cf_def.key_cache_size == null ? CFMetaData.DEFAULT_KEY_CACHE_SIZE : cf_def.key_cache_size, - cf_def.read_repair_chance == null ? CFMetaData.DEFAULT_READ_REPAIR_CHANCE : cf_def.read_repair_chance, - cf_def.replicate_on_write == null ? CFMetaData.DEFAULT_REPLICATE_ON_WRITE : cf_def.replicate_on_write, - cf_def.gc_grace_seconds != null ? cf_def.gc_grace_seconds : CFMetaData.DEFAULT_GC_GRACE_SECONDS, - DatabaseDescriptor.getComparator(validate), - cf_def.min_compaction_threshold == null ? CFMetaData.DEFAULT_MIN_COMPACTION_THRESHOLD : cf_def.min_compaction_threshold, - cf_def.max_compaction_threshold == null ? CFMetaData.DEFAULT_MAX_COMPACTION_THRESHOLD : cf_def.max_compaction_threshold, - cf_def.row_cache_save_period_in_seconds == null ? CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS : cf_def.row_cache_save_period_in_seconds, - cf_def.key_cache_save_period_in_seconds == null ? CFMetaData.DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS : cf_def.key_cache_save_period_in_seconds, - cf_def.memtable_flush_after_mins == null ? CFMetaData.DEFAULT_MEMTABLE_LIFETIME_IN_MINS : cf_def.memtable_flush_after_mins, - cf_def.memtable_throughput_in_mb == null ? CFMetaData.DEFAULT_MEMTABLE_THROUGHPUT_IN_MB : cf_def.memtable_throughput_in_mb, - cf_def.memtable_operations_in_millions == null ? CFMetaData.DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS : cf_def.memtable_operations_in_millions, - ColumnDefinition.fromColumnDefs((Iterable) cf_def.column_metadata)); - } - - public KsDef describe_keyspace(CharSequence keyspace) throws AvroRemoteException, NotFoundException - { - KSMetaData ksMetadata = DatabaseDescriptor.getTableDefinition(keyspace.toString()); - if (ksMetadata == null) - throw new NotFoundException(); - - KsDef ksDef = new KsDef(); - ksDef.name = keyspace; - ksDef.replication_factor = ksMetadata.replicationFactor; - ksDef.strategy_class = ksMetadata.strategyClass.getName(); - if (ksMetadata.strategyOptions != null) - { - ksDef.strategy_options = new HashMap(); - ksDef.strategy_options.putAll(ksMetadata.strategyOptions); - } - - GenericArray cfDefs = new GenericData.Array(ksMetadata.cfMetaData().size(), Schema.createArray(CfDef.SCHEMA$)); - for (CFMetaData cfm : ksMetadata.cfMetaData().values()) - { - cfDefs.add(CFMetaData.convertToAvro(cfm)); - } - ksDef.cf_defs = cfDefs; - - return ksDef; - } - - public CharSequence system_drop_column_family(CharSequence column_family) throws AvroRemoteException, InvalidRequestException - { - checkKeyspaceAndLoginAuthorized(Permission.WRITE); - - try - { - applyMigrationOnStage(new DropColumnFamily(state().getKeyspace(), column_family.toString())); - return DatabaseDescriptor.getDefsVersion().toString(); - } - catch (ConfigurationException e) - { - throw newInvalidRequestException(e); - } - catch (IOException e) - { - throw newInvalidRequestException(e); - } - } - - public CharSequence system_drop_keyspace(CharSequence keyspace) throws AvroRemoteException, InvalidRequestException - { - if (!(DatabaseDescriptor.getAuthenticator() instanceof AllowAllAuthenticator)) - throw newInvalidRequestException("Unable to create new keyspace while authentication is enabled."); - - try - { - applyMigrationOnStage(new DropKeyspace(keyspace.toString())); - return DatabaseDescriptor.getDefsVersion().toString(); - } - catch (ConfigurationException e) - { - throw newInvalidRequestException(e); - } - catch (IOException e) - { - throw newInvalidRequestException(e); - } - } - - public CharSequence describe_partitioner() throws AvroRemoteException - { - return StorageService.getPartitioner().getClass().getName(); - } - - public List describe_splits(CharSequence cfName, CharSequence start_token, CharSequence end_token, int keys_per_split) { - Token.TokenFactory tf = StorageService.getPartitioner().getTokenFactory(); - List tokens = StorageService.instance.getSplits(state().getKeyspace(), cfName.toString(), new Range(tf.fromString(start_token.toString()), tf.fromString(end_token.toString())), keys_per_split); - List splits = new ArrayList(tokens.size()); - for (Token token : tokens) - { - splits.add(tf.toString(token)); - } - return splits; - } - - public List multiget_count(List keys, ColumnParent columnParent, SlicePredicate predicate, ConsistencyLevel consistencyLevel) - throws AvroRemoteException, InvalidRequestException, UnavailableException, TimedOutException - { - if (logger.isDebugEnabled()) - logger.debug("multiget_count"); - - checkKeyspaceAndLoginAuthorized(Permission.READ); - String keyspace = state().getKeyspace(); - - List counts = new ArrayList(); - List columnFamiliesMap = multigetSliceInternal(keyspace, keys, columnParent, predicate, consistencyLevel); - - for (CoscsMapEntry cf : columnFamiliesMap) - { - KeyCountMapEntry countEntry = new KeyCountMapEntry(); - countEntry.key = cf.key; - countEntry.count = cf.columns.size(); - counts.add(countEntry); - } - - return counts; - } - - public List describe_ring(CharSequence keyspace) throws AvroRemoteException, InvalidRequestException - { - if (keyspace == null || keyspace.toString().equals(Table.SYSTEM_TABLE)) - throw newInvalidRequestException("There is no ring for the keyspace: " + keyspace); - List ranges = new ArrayList(); - Token.TokenFactory tf = StorageService.getPartitioner().getTokenFactory(); - for (Map.Entry> entry : StorageService.instance.getRangeToEndpointMap(keyspace.toString()).entrySet()) - { - Range range = entry.getKey(); - List endpoints = entry.getValue(); - ranges.add(newTokenRange(tf.toString(range.left), tf.toString(range.right), endpoints)); - } - return ranges; - } - - public Void truncate(CharSequence columnFamily) throws AvroRemoteException, InvalidRequestException, UnavailableException - { - if (logger.isDebugEnabled()) - logger.debug("truncating {} in {}", columnFamily, state().getKeyspace()); - - try - { - state().hasColumnFamilyAccess(columnFamily.toString(), Permission.WRITE); - schedule(); - StorageProxy.truncateBlocking(state().getKeyspace(), columnFamily.toString()); - } - catch (org.apache.cassandra.thrift.InvalidRequestException e) - { - throw newInvalidRequestException(e); - } - catch (org.apache.cassandra.thrift.UnavailableException e) - { - throw newUnavailableException(e); - } - catch (TimeoutException e) - { - throw newUnavailableException(e); - } - catch (IOException e) - { - throw newUnavailableException(e); - } - finally - { - release(); - } - return null; - } - - public List get_range_slices(ColumnParent column_parent, SlicePredicate slice_predicate, KeyRange range, ConsistencyLevel consistency_level) - throws InvalidRequestException, TimedOutException - { - String keyspace = state().getKeyspace(); - try - { - state().hasColumnFamilyAccess(column_parent.column_family.toString(), Permission.READ); - } - catch (org.apache.cassandra.thrift.InvalidRequestException thriftE) - { - throw newInvalidRequestException(thriftE); - } - - AvroValidation.validateColumnParent(keyspace, column_parent); - AvroValidation.validatePredicate(keyspace, column_parent, slice_predicate); - AvroValidation.validateKeyRange(range); - - List rows; - try - { - IPartitioner p = StorageService.getPartitioner(); - AbstractBounds bounds; - if (range.start_key == null) - { - Token.TokenFactory tokenFactory = p.getTokenFactory(); - Token left = tokenFactory.fromString(range.start_token.toString()); - Token right = tokenFactory.fromString(range.end_token.toString()); - bounds = new Range(left, right); - } - else - { - bounds = new Bounds(p.getToken(range.start_key), p.getToken(range.end_key)); - } - try - { - schedule(); - rows = StorageProxy.getRangeSlice(new RangeSliceCommand(keyspace, - thriftColumnParent(column_parent), - thriftSlicePredicate(slice_predicate), - bounds, - range.count), - thriftConsistencyLevel(consistency_level)); - } - catch (org.apache.cassandra.thrift.UnavailableException thriftE) - { - throw newUnavailableException(thriftE); - } - finally - { - release(); - } - assert rows != null; - } - catch (TimeoutException e) - { - throw new TimedOutException(); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - return avronateKeySlices(rows, column_parent, slice_predicate); - } - - public List get_indexed_slices(ColumnParent column_parent, IndexClause index_clause, SlicePredicate column_predicate, ConsistencyLevel consistency_level) - throws InvalidRequestException, UnavailableException, TimedOutException - { - if (logger.isDebugEnabled()) - logger.debug("scan"); - - try - { - state().hasColumnFamilyAccess(column_parent.column_family.toString(), Permission.READ); - } - catch (org.apache.cassandra.thrift.InvalidRequestException thriftE) - { - throw newInvalidRequestException(thriftE); - } - - String keyspace = state().getKeyspace(); - AvroValidation.validateColumnParent(keyspace, column_parent); - AvroValidation.validatePredicate(keyspace, column_parent, column_predicate); - AvroValidation.validateIndexClauses(keyspace, column_parent.column_family.toString(), index_clause); - - List rows; - try - { - rows = StorageProxy.scan(keyspace.toString(), - column_parent.column_family.toString(), - thriftIndexClause(index_clause), - thriftSlicePredicate(column_predicate), - thriftConsistencyLevel(consistency_level)); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - catch (TimeoutException e) - { - throw new TimedOutException(); - } - catch (org.apache.cassandra.thrift.UnavailableException e) - { - throw newUnavailableException(); - } - return avronateKeySlices(rows, column_parent, column_predicate); - } - - private List avronateKeySlices(List rows, ColumnParent column_parent, SlicePredicate predicate) - { - List keySlices = new ArrayList(rows.size()); - boolean reversed = predicate.slice_range != null && predicate.slice_range.reversed; - for (Row row : rows) - { - List avronatedColumns = avronateColumnFamily(row.cf, column_parent.super_column != null, reversed); - keySlices.add(newKeySlice(row.key.key, avronatedColumns)); - } - - return keySlices; - } - - private org.apache.cassandra.thrift.ColumnParent thriftColumnParent(ColumnParent avro_column_parent) - { - org.apache.cassandra.thrift.ColumnParent cp = new org.apache.cassandra.thrift.ColumnParent(avro_column_parent.column_family.toString()); - if (avro_column_parent.super_column != null) - cp.super_column = avro_column_parent.super_column; - - return cp; - } - - private org.apache.cassandra.thrift.SlicePredicate thriftSlicePredicate(SlicePredicate avro_pred) { - // One or the other are set, so check for nulls of either - org.apache.cassandra.thrift.SliceRange slice_range = (avro_pred.slice_range != null) - ? thriftSliceRange(avro_pred.slice_range) - : null; - - return new org.apache.cassandra.thrift.SlicePredicate().setColumn_names(avro_pred.column_names).setSlice_range(slice_range); - } - - private org.apache.cassandra.thrift.SliceRange thriftSliceRange(SliceRange avro_range) { - return new org.apache.cassandra.thrift.SliceRange(avro_range.start, avro_range.finish, avro_range.reversed, avro_range.count); - } - - private org.apache.cassandra.thrift.IndexClause thriftIndexClause(IndexClause avro_clause) { - List expressions = new ArrayList(); - for(IndexExpression exp : avro_clause.expressions) - expressions.add(thriftIndexExpression(exp)); - - return new org.apache.cassandra.thrift.IndexClause(expressions, avro_clause.start_key, avro_clause.count); - } - - private org.apache.cassandra.thrift.IndexExpression thriftIndexExpression(IndexExpression avro_exp) { - return new org.apache.cassandra.thrift.IndexExpression(avro_exp.column_name, thriftIndexOperator(avro_exp.op), avro_exp.value); - } - - private org.apache.cassandra.thrift.IndexOperator thriftIndexOperator(IndexOperator avro_op) { - switch (avro_op) - { - case EQ: return org.apache.cassandra.thrift.IndexOperator.EQ; - case GTE: return org.apache.cassandra.thrift.IndexOperator.GTE; - case GT: return org.apache.cassandra.thrift.IndexOperator.GT; - case LTE: return org.apache.cassandra.thrift.IndexOperator.LTE; - case LT: return org.apache.cassandra.thrift.IndexOperator.LT; - } - return null; - } -} diff --git a/src/java/org/apache/cassandra/avro/KeyspaceNotDefinedException.java b/src/java/org/apache/cassandra/avro/KeyspaceNotDefinedException.java deleted file mode 100644 index a4a9e108aa..0000000000 --- a/src/java/org/apache/cassandra/avro/KeyspaceNotDefinedException.java +++ /dev/null @@ -1,34 +0,0 @@ -package org.apache.cassandra.avro; -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ - - -import org.apache.avro.util.Utf8; - -// XXX: This is an analogue to org.apache.cassandra.db.KeyspaceNotDefinedException -@SuppressWarnings("serial") -public class KeyspaceNotDefinedException extends InvalidRequestException { - - public KeyspaceNotDefinedException(Utf8 why) - { - this.why = why; - } -}