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

merge from 0.8 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1092687 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Dusbabek 2011-04-15 12:48:15 +00:00
parent 842f141391
commit 8a6feefe8f
33 changed files with 2261 additions and 1127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGES.txt
View File

@ -1,7 +1,8 @@
0.8-dev
* remove Avro RPC support (CASSANDRA-926)
* adds support for columns that act as incr/decr counters
(CASSANDRA-1072, 1937, 1944, 1936, 2101, 2093, 2288, 2105, 2384, 2236, 2342)
(CASSANDRA-1072, 1937, 1944, 1936, 2101, 2093, 2288, 2105, 2384, 2236, 2342,
2454)
* CQL (CASSANDRA-1703, 1704, 1705, 1706, 1707, 1708, 1710, 1711, 1940,
2124, 2302, 2277)
* avoid double RowMutation serialization on write path (CASSANDRA-1800)
@ -23,6 +24,7 @@
* compaction throttling (CASSANDRA-2156)
* add key type information and alias (CASSANDRA-2311, 2396)
* cli no longer divides read_repair_chance by 100 (CASSANDRA-2458)
* made CompactionInfo.getTaskType return an enum (CASSANDRA-2482)
0.7.5

5
build.xml
View File

@ -27,6 +27,7 @@
<property name="basedir" value="."/>
<property name="build.src" value="${basedir}/src"/>
<property name="build.src.java" value="${basedir}/src/java"/>
<property name="build.src.resources" value="${basedir}/src/resources"/>
<property name="build.src.driver" value="${basedir}/drivers/java/src" />
<property name="avro.src" value="${basedir}/src/avro"/>
<property name="build.src.gen-java" value="${basedir}/src/gen-java"/>
@ -383,7 +384,9 @@
<src path="${build.src.driver}" />
<classpath refid="cassandra.classpath"/>
</javac>
<copy todir="${build.classes.main}">
<fileset dir="${build.src.resources}" />
</copy>
<taskdef name="paranamer" classname="com.thoughtworks.paranamer.ant.ParanamerGeneratorTask">
<classpath refid="cassandra.classpath" />
</taskdef>

30
drivers/py/README
View File

@ -1,2 +1,28 @@
Python language driver for CQL, a SQL-alike query language for Apache
Cassandra.
A Python driver for CQL that adheres to py-dbapi v2
(PEP249, Python Database API Specification v2.0: http://www.python.org/dev/peps/pep-0249/).
Standard use:
>> import cql
>> con = cql.connect(host, port, keyspace)
>> cursor = con.cursor()
>> cursor.execute("CQL QUERY", {kw=Foo, kw2=Bar, etc...})
- cursor.description # None initially, list of N tuples that represent
the N columns in a row after an execute. Only
contains type and name info, not values.
- cursor.rowcount # -1 initially, N after an execute
- cursor.arraysize # variable size of a fetchmany call
- cursor.fetchone() # returns a single row
- cursor.fetchmany() # returns self.arraysize # of rows
- cursor.fetchall() # returns all rows, don't do this.
>> cursor.execute("ANOTHER QUERY", **more_kwargs)
>> for row in cursor: # Iteration is equivalent to lots of fetchone() calls
>> doRowMagic(row)
>> cursor.close()
>> con.close()
Query substitution:
- Use named parameters and a dictionary of names and values.
e.g. execute("SELECT * FROM CF WHERE name=:name", name="Foo")

83
drivers/py/cql/__init__.py
View File

@ -15,9 +15,82 @@
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Cassandra Query Language driver
"""
import exceptions
import datetime
import connection
import marshal
# dbapi Error hierarchy
class Warning(exceptions.StandardError): pass
class Error (exceptions.StandardError): pass
class InterfaceError(Error): pass
class DatabaseError (Error): pass
class DataError (DatabaseError): pass
class OperationalError (DatabaseError): pass
class IntegrityError (DatabaseError): pass
class InternalError (DatabaseError): pass
class ProgrammingError (DatabaseError): pass
class NotSupportedError(DatabaseError): pass
# Module constants
apilevel = 1.0
threadsafety = 1 # Threads may share the module, but not connections/cursors.
paramstyle = 'named'
ROW_KEY = "Row Key"
# TODO: Pull connections out of a pool instead.
def connect(host, port=9160, keyspace='system', user=None, password=None):
return connection.Connection(host, port, keyspace, user, password)
# Module Type Objects and Constructors
Date = datetime.date
Time = datetime.time
Timestamp = datetime.datetime
Binary = buffer
def DateFromTicks(ticks):
return Date(*time.localtime(ticks)[:3])
def TimeFromTicks(ticks):
return Time(*time.localtime(ticks)[3:6])
def TimestampFromTicks(ticks):
return Timestamp(*time.localtime(ticks)[:6])
class DBAPITypeObject:
def __init__(self, *values):
self.values = values
def __cmp__(self,other):
if other in self.values:
return 0
if other < self.values:
return 1
else:
return -1
STRING = DBAPITypeObject(marshal.BYTES_TYPE, marshal.ASCII_TYPE, marshal.UTF8_TYPE)
BINARY = DBAPITypeObject(marshal.BYTES_TYPE, marshal.UUID_TYPE, marshal.LEXICAL_UUID_TYPE)
NUMBER = DBAPITypeObject(marshal.LONG_TYPE, marshal.INTEGER_TYPE)
DATETIME = DBAPITypeObject(marshal.TIME_UUID_TYPE)
ROWID = DBAPITypeObject(marshal.BYTES_TYPE, marshal.ASCII_TYPE, marshal.UTF8_TYPE,
marshal.INTEGER_TYPE, marshal.LONG_TYPE, marshal.UUID_TYPE,
marshal.LEXICAL_UUID_TYPE, marshal.TIME_UUID_TYPE)
from connection import Connection
from connection_pool import ConnectionPool

194
drivers/py/cql/connection.py
View File

@ -15,183 +15,69 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from os.path import exists, abspath, dirname, join
from cursor import Cursor
from cassandra import Cassandra
from thrift.transport import TTransport, TSocket
from thrift.protocol import TBinaryProtocol
from thrift.Thrift import TApplicationException
from errors import CQLException, InvalidCompressionScheme
from marshal import prepare
from decoders import SchemaDecoder
from results import RowsProxy
import zlib, re
from cassandra import Cassandra
from cassandra.ttypes import Compression, InvalidRequestException, \
CqlResultType, AuthenticationRequest, \
SchemaDisagreementException
COMPRESSION_SCHEMES = ['GZIP', 'NONE']
DEFAULT_COMPRESSION = 'GZIP'
__all__ = ['COMPRESSION_SCHEMES', 'DEFAULT_COMPRESSION', 'Connection']
class Connection(object):
"""
CQL connection object.
Example usage:
>>> conn = Connection("localhost", keyspace="Keyspace1")
>>> r = conn.execute('SELECT "age" FROM Users')
>>> for row in r.rows:
... for column in row.columns:
... print "%s is %s years of age" % (r.key, column.age)
"""
_keyspace_re = re.compile("USE (\w+);?", re.I | re.M)
_cfamily_re = re.compile("\s*SELECT\s+.+\s+FROM\s+[\']?(\w+)", re.I | re.M)
_ddl_re = re.compile("\s*(CREATE|ALTER|DROP)\s+", re.I | re.M)
def __init__(self, host, port=9160, keyspace=None, username=None,
password=None, decoder=None):
def __init__(self, host, port, keyspace, user=None, password=None):
"""
Params:
* host .........: hostname of Cassandra node.
* port .........: port number to connect to (optional).
* keyspace .....: keyspace name (optional).
* username .....: username used in authentication (optional).
* port .........: port number to connect to.
* keyspace .....: keyspace to connect to.
* user .........: username used in authentication (optional).
* password .....: password used in authentication (optional).
* decoder ......: result decoder instance (optional, defaults to none).
"""
self.host = host
self.port = port
self.keyspace = keyspace
socket = TSocket.TSocket(host, port)
self.transport = TTransport.TFramedTransport(socket)
protocol = TBinaryProtocol.TBinaryProtocolAccelerated(self.transport)
self.client = Cassandra.Client(protocol)
socket.open()
# XXX: "current" is probably a misnomer.
self._cur_keyspace = None
self._cur_column_family = None
if username and password:
credentials = {"username": username, "password": password}
self.open_socket = True
if user and password:
credentials = {"username": user, "password": password}
self.client.login(AuthenticationRequest(credentials=credentials))
if keyspace:
self.execute('USE %s;' % keyspace)
self._cur_keyspace = keyspace
if not decoder:
self.decoder = SchemaDecoder(self.__get_schema())
else:
self.decoder = decoder
c = self.cursor()
c.execute('USE %s;' % keyspace)
c.close()
def __get_schema(self):
def columns(metadata):
results = {}
for col in metadata:
results[col.name] = col.validation_class
return results
def column_families(cf_defs):
cfresults = {}
for cf in cf_defs:
cfresults[cf.name] = {"comparator": cf.comparator_type}
cfresults[cf.name]["default_validation_class"] = \
cf.default_validation_class
cfresults[cf.name]["key_validation_class"] = \
cf.key_validation_class
cfresults[cf.name]["columns"] = columns(cf.column_metadata)
return cfresults
schema = {}
for ksdef in self.client.describe_keyspaces():
schema[ksdef.name] = column_families(ksdef.cf_defs)
return schema
def prepare(self, query, *args):
prepared_query = prepare(query, *args)
# Snag the keyspace or column family and stash it for later use in
# decoding columns. These regexes don't match every query, but the
# current column family only needs to be current for SELECTs.
match = Connection._cfamily_re.match(prepared_query)
if match:
self._cur_column_family = match.group(1)
return prepared_query
match = Connection._keyspace_re.match(prepared_query)
if match:
self._cur_keyspace = match.group(1)
return prepared_query
# If this is a CREATE, then refresh the schema for decoding purposes.
match = Connection._ddl_re.match(prepared_query)
if match:
if isinstance(self.decoder, SchemaDecoder):
self.decoder.schema = self.__get_schema()
return prepared_query
def __str__(self):
return "{host: '%s:%s', keyspace: '%s'}"%(self.host,self.port,self.keyspace)
def execute(self, query, *args, **kwargs):
"""
Execute a CQL query on a remote node.
Params:
* query .........: CQL query string.
* args ..........: Query parameters.
* compression ...: Query compression type (optional).
"""
if kwargs.has_key("compression"):
compress = kwargs.get("compression").upper()
else:
compress = DEFAULT_COMPRESSION
compressed_query = Connection.compress_query(self.prepare(query, *args),
compress)
request_compression = getattr(Compression, compress)
try:
response = self.client.execute_cql_query(compressed_query,
request_compression)
except InvalidRequestException, ire:
raise CQLException("Bad Request: %s" % ire.why)
except SchemaDisagreementException, sde:
raise CQLException("schema versions disagree, (try again later).")
except TApplicationException, tapp:
raise CQLException("Internal application error")
except Exception, exc:
raise CQLException(exc)
if response.type == CqlResultType.ROWS:
return RowsProxy(response.rows,
self._cur_keyspace,
self._cur_column_family,
self.decoder)
if response.type == CqlResultType.INT:
return response.num
return None
###
# Connection API
###
def close(self):
if not self.open_socket:
raise InternalError("Connection has been closed.")
self.transport.close()
def is_open(self):
return self.transport.isOpen()
self.open_socket = False
@classmethod
def compress_query(cls, query, compression):
def commit(self):
"""
Returns a query string compressed with the specified compression type.
Params:
* query .........: The query string to compress.
* compression ...: Type of compression to use.
'Database modules that do not support transactions should
implement this method with void functionality.'
"""
if not compression in COMPRESSION_SCHEMES:
raise InvalidCompressionScheme(compression)
return
if compression == 'GZIP':
return zlib.compress(query)
return query
# vi: ai ts=4 tw=0 sw=4 et
def rollback(self):
raise NotSupportedError("Rollback functionality not present in Cassandra.")
def cursor(self):
if not self.open_socket:
raise InternalError("Connection has been closed.")
return Cursor(self)

225
drivers/py/cql/cursor.py Normal file
View File

@ -0,0 +1,225 @@
# 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 re
import zlib
import cql
from cql.marshal import prepare
from cql.decoders import SchemaDecoder
from cql.results import ResultSet
from cql.cassandra.ttypes import (Compression, CqlResultType, InvalidRequestException,
TApplicationException)
class Cursor:
_keyspace_re = re.compile("USE (\w+);?", re.I | re.M)
_cfamily_re = re.compile("\s*SELECT\s+.+\s+FROM\s+[\']?(\w+)", re.I | re.M)
_ddl_re = re.compile("\s*(CREATE|ALTER|DROP)\s+", re.I | re.M)
def __init__(self, parent_connection):
self.open_socket = True
self.parent_connection = parent_connection
self.result = None # Populate on execute()
self.description = None # A list of 7-tuples:
# (column_name, type_code, none, none,
# none, none, nulls_ok=True)
# Populate on execute()
self.arraysize = 1
self.rowcount = -1 # Populate on execute()
self.compression = 'GZIP'
self._query_ks = self.parent_connection.keyspace
self._query_cf = None
self.decoder = SchemaDecoder(self.__get_schema())
###
# Cursor API
###
def close(self):
self.open_socket = False
def prepare(self, query, params):
prepared_query = prepare(query, params)
# Snag the keyspace or column family and stash it for later use in
# decoding columns. These regexes don't match every query, but the
# current column family only needs to be current for SELECTs.
match = Cursor._cfamily_re.match(prepared_query)
if match:
self._query_cf = match.group(1)
return prepared_query
match = Cursor._keyspace_re.match(prepared_query)
if match:
self._query_ks = match.group(1)
return prepared_query
# If this is a CREATE, then refresh the schema for decoding purposes.
match = Cursor._ddl_re.match(prepared_query)
if match:
if isinstance(self.decoder, SchemaDecoder):
self.decoder.schema = self.__get_schema()
return prepared_query
def __get_schema(self):
def columns(metadata):
results = {}
for col in metadata:
results[col.name] = col.validation_class
return results
def column_families(cf_defs):
cfresults = {}
if cf_defs:
for cf in cf_defs:
cfresults[cf.name] = {"comparator": cf.comparator_type}
cfresults[cf.name]["default_validation_class"] = \
cf.default_validation_class
cfresults[cf.name]["key_validation_class"] = \
cf.key_validation_class
cfresults[cf.name]["columns"] = columns(cf.column_metadata)
return cfresults
schema = {}
client = self.parent_connection.client
for ksdef in client.describe_keyspaces():
schema[ksdef.name] = column_families(ksdef.cf_defs)
return schema
def execute(self, cql_query, params={}):
self.__checksock()
try:
prepared_q = self.prepare(cql_query, params)
except KeyError, e:
raise cql.ProgrammingError("Unmatched named substitution: " +
"%s not given for %s" % (e, cql_query))
if self.compression == 'GZIP':
compressed_q = zlib.compress(prepared_q)
else:
compressed_q = prepared_q
request_compression = getattr(Compression, self.compression)
try:
client = self.parent_connection.client
response = client.execute_cql_query(compressed_q, request_compression)
except InvalidRequestException, ire:
raise cql.ProgrammingError("Bad Request: %s" % ire.why)
except SchemaDisagreementException, sde:
raise cql.IntegrityError("Schema versions disagree, (try again later).")
except TApplicationException, tapp:
raise cql.InternalError("Internal application error")
if response.type == CqlResultType.ROWS:
self.result = ResultSet(response.rows,
self._query_ks,
self._query_cf,
self.decoder)
self.rs_idx = 0
self.rowcount = len(self.result)
self.description = self.result.description
if response.type == CqlResultType.INT:
self.result = [(response.num,)]
self.rs_idx = 0
self.rowcount = 1
# TODO: name could be the COUNT expression
self.description = (None, None, None, None, None, None, None)
# 'Return values are not defined.'
return True
def executemany(self, operation_list, argslist):
self.__checksock()
opssize = len(operation_list)
argsize = len(argslist)
if opssize > argsize:
raise cql.InterfaceError("Operations outnumber args for executemany().")
elif opssize < argsize:
raise cql.InterfaceError("Args outnumber operations for executemany().")
for idx in xrange(opssize):
self.execute(operation_list[idx], *argslist[idx])
def fetchone(self):
self.__checksock()
ret = self.result[self.rs_idx]
self.rs_idx += 1
self.description = getattr(self.result, 'description', self.description)
return ret
def fetchmany(self, size=None):
self.__checksock()
if size is None:
size = self.arraysize
end = self.rs_idx + size
ret = self.result[self.rs_idx:end]
self.rs_idx = end
self.description = getattr(self.result, 'description', self.description)
return ret
def fetchall(self):
self.__checksock()
ret = self.result[self.rs_idx:]
self.rs_idx = len(self.result)
self.description = self.result.description
return ret
###
# Iterator extension
###
def next(self):
raise Warning("DB-API extension cursor.next() used")
if self.rs_idx >= len(self.result):
raise StopIteration
return self.fetchone()
def __iter__(self):
raise Warning("DB-API extension cursor.__iter__() used")
return self
###
# Unsupported, unimplemented optionally
###
def setinputsizes(self, sizes):
pass # DO NOTHING
def setoutputsize(self, size, *columns):
pass # DO NOTHING
def callproc(self, procname, *args):
raise cql.NotSupportedError()
def nextset(self):
raise cql.NotSupportedError()
###
# Helpers
###
def __checksock(self):
if not self.open_socket:
raise cql.InternalError("Cursor belonging to %s has been closed." %
(self.parent_connection, ))

59
drivers/py/cql/decoders.py
View File

@ -15,62 +15,53 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from os.path import abspath, dirname, join
from marshal import unmarshal
import cql
from marshal import (unmarshallers, unmarshal_noop)
class BaseDecoder(object):
def decode_column(self, keyspace, column_family, name, value):
raise NotImplementedError()
def decode_key(self, keyspace, column_family, key):
raise NotImplementedError()
class NoopDecoder(BaseDecoder):
def decode_column(self, keyspace, column_family, name, value):
return (name, value)
def decode_key(self, keyspace, column_family, key):
return key
class SchemaDecoder(BaseDecoder):
class SchemaDecoder(object):
"""
Decode binary column names/values according to schema.
"""
def __init__(self, schema={}):
self.schema = schema
def __get_column_family_def(self, keyspace, column_family):
if self.schema.has_key(keyspace):
if self.schema[keyspace].has_key(column_family):
return self.schema[keyspace][column_family]
return None
def __comparator_for(self, keyspace, column_family):
cfam = self.__get_column_family_def(keyspace, column_family)
if cfam and cfam.has_key("comparator"):
if cfam and "comparator" in cfam:
return cfam["comparator"]
return None
def __validator_for(self, keyspace, column_family, name):
cfam = self.__get_column_family_def(keyspace, column_family)
if cfam:
if cfam["columns"].has_key(name):
if name in cfam["columns"]:
return cfam["columns"][name]
else:
return cfam["default_validation_class"]
return cfam["default_validation_class"]
return None
def __keytype_for(self, keyspace, column_family, key):
def __keytype_for(self, keyspace, column_family):
cfam = self.__get_column_family_def(keyspace, column_family)
if cfam and cfam.has_key("key_validation_class"):
if cfam and "key_validation_class" in cfam:
return cfam["key_validation_class"]
return None
def decode_column(self, keyspace, column_family, name, value):
def decode_row(self, keyspace, column_family, row):
key_type = self.__keytype_for(keyspace, column_family)
key = unmarshallers.get(key_type, unmarshal_noop)(row.key)
description = [(cql.ROW_KEY, key_type, None, None, None, None, None, False)]
comparator = self.__comparator_for(keyspace, column_family)
validator = self.__validator_for(keyspace, column_family, name)
return (unmarshal(name, comparator), unmarshal(value, validator))
def decode_key(self, keyspace, column_family, key):
return unmarshal(key, self.__keytype_for(keyspace, column_family, key))
unmarshal = unmarshallers.get(comparator, unmarshal_noop)
values = [key]
for column in row.columns:
description.append((unmarshal(column.name), comparator, None, None, None, None, True))
validator = self.__validator_for(keyspace, column_family, column.name)
values.append(unmarshallers.get(validator, unmarshal_noop)(column.value))
return description, values

3
drivers/py/cql/errors.py
View File

@ -15,8 +15,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
__all__ = ['InvalidCompressionScheme', 'CQLException', 'InvalidQueryFormat']
__all__ = ['InvalidCompressionScheme', 'InvalidQueryFormat']
class InvalidCompressionScheme(Exception): pass
class InvalidQueryFormat(Exception): pass
class CQLException(Exception): pass

109
drivers/py/cql/marshal.py
View File

@ -15,68 +15,75 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import re
import struct
from uuid import UUID
from StringIO import StringIO
from errors import InvalidQueryFormat
from struct import unpack
__all__ = ['prepare', 'marshal', 'unmarshal']
import cql
def prepare(query, *args):
result = StringIO()
index = query.find('?')
oldindex = 0
count = 0
while (index >= 0):
result.write(query[oldindex:index])
try:
result.write(marshal(args[count]))
except IndexError:
raise InvalidQueryFormat("not enough arguments in substitution")
oldindex = index + 1
index = query.find('?', index + 1)
count += 1
result.write(query[oldindex:])
if count < len(args):
raise InvalidQueryFormat("too many arguments in substitution")
return result.getvalue()
__all__ = ['prepare', 'marshal', 'unmarshal_noop', 'unmarshallers']
if hasattr(struct, 'Struct'): # new in Python 2.5
_have_struct = True
_long_packer = struct.Struct('>q')
else:
_have_struct = False
_param_re = re.compile(r"(?<!strategy_options)(:[a-zA-Z_][a-zA-Z0-9_]*)", re.M)
BYTES_TYPE = "org.apache.cassandra.db.marshal.BytesType"
ASCII_TYPE = "org.apache.cassandra.db.marshal.AsciiType"
UTF8_TYPE = "org.apache.cassandra.db.marshal.UTF8Type"
INTEGER_TYPE = "org.apache.cassandra.db.marshal.IntegerType"
LONG_TYPE = "org.apache.cassandra.db.marshal.LongType"
UUID_TYPE = "org.apache.cassandra.db.marshal.UUIDType"
LEXICAL_UUID_TYPE = "org.apache.cassandra.db.marshal.LexicalType"
TIME_UUID_TYPE = "org.apache.cassandra.db.marshal.TimeUUIDType"
def prepare(query, params):
# For every match of the form ":param_name", call marshal
# on kwargs['param_name'] and replace that section of the query
# with the result
new, count = re.subn(_param_re, lambda m: marshal(params[m.group(1)[1:]]), query)
if len(params) > count:
raise cql.ProgrammingError("More keywords were provided than parameters")
return new
def marshal(term):
if isinstance(term, (long,int)):
return "%d" % term
elif isinstance(term, unicode):
if isinstance(term, unicode):
return "'%s'" % __escape_quotes(term.encode('utf8'))
elif isinstance(term, str):
return "'%s'" % __escape_quotes(term)
elif isinstance(term, UUID):
else:
return str(term)
def unmarshal_noop(bytestr):
return bytestr
def unmarshal_utf8(bytestr):
return bytestr.decode("utf8")
def unmarshal_int(bytestr):
return decode_bigint(bytestr)
def unmarshal_long(bytestr):
if _have_struct:
return _long_packer.unpack(bytestr)[0]
else:
return str(term)
def unmarshal(bytestr, typestr):
if typestr == "org.apache.cassandra.db.marshal.BytesType":
return bytestr
elif typestr == "org.apache.cassandra.db.marshal.AsciiType":
return bytestr
elif typestr == "org.apache.cassandra.db.marshal.UTF8Type":
return bytestr.decode("utf8")
elif typestr == "org.apache.cassandra.db.marshal.IntegerType":
return decode_bigint(bytestr)
elif typestr == "org.apache.cassandra.db.marshal.LongType":
return unpack(">q", bytestr)[0]
elif typestr == "org.apache.cassandra.db.marshal.UUIDType":
return UUID(bytes=bytestr)
elif typestr == "org.apache.cassandra.db.marshal.LexicalUUIDType":
return UUID(bytes=bytestr)
elif typestr == "org.apache.cassandra.db.marshal.TimeUUIDType":
return UUID(bytes=bytestr)
else:
return bytestr
def unmarshal_uuid(bytestr):
return UUID(bytes=bytestr)
unmarshallers = {BYTES_TYPE: unmarshal_noop,
ASCII_TYPE: unmarshal_noop,
UTF8_TYPE: unmarshal_utf8,
INTEGER_TYPE: unmarshal_int,
LONG_TYPE: unmarshal_long,
UUID_TYPE: unmarshal_uuid,
LEXICAL_UUID_TYPE: unmarshal_uuid,
TIME_UUID_TYPE: unmarshal_uuid}
def decode_bigint(term):
val = int(term.encode('hex'), 16)
if (ord(term[0]) & 128) != 0:

110
drivers/py/cql/results.py
View File

@ -15,84 +15,48 @@
# See the License for the specific language governing permissions and
# limitations under the License.
class RowsProxy(object):
def __init__(self, rows, keyspace, cfam, decoder):
class ResultSet(object):
def __init__(self, rows, keyspace, column_family, decoder):
self.rows = rows
self.keyspace = keyspace
self.cfam = cfam
self.ks = keyspace
self.cf = column_family
self.decoder = decoder
# We need to try to parse the first row to set the description
if len(self.rows) > 0:
self.description, self._first_vals = self.decoder.decode_row(self.ks, self.cf, self.rows[0])
else:
self.description, self._first_vals = (None, None)
def __len__(self):
return len(self.rows)
def __getitem__(self, idx):
return Row(self.rows[idx].key,
self.rows[idx].columns,
self.keyspace,
self.cfam,
self.decoder)
if isinstance(idx, int):
if idx == 0 and self._first_vals:
return self._first_vals
self.description, vals = self.decoder.decode_row(self.ks, self.cf, self.rows[idx])
return vals
elif isinstance(idx, slice):
num_rows = len(self.rows)
results = []
for i in xrange(idx.start, min(len(self.rows), idx.stop)):
if i == 0 and self._first_vals:
vals = self._first_vals
else:
self.description, vals = self.decoder.decode_row(self.ks, self.cf, self.rows[i])
results.append(vals)
return results
else:
raise TypeError
def __iter__(self):
idx = 0
for r in self.rows:
yield Row(r.key, r.columns, self.keyspace, self.cfam, self.decoder)
class Row(object):
def __init__(self, key, columns, keyspace, cfam, decoder):
self._key = key
self.keyspace = keyspace
self.cfam = cfam
self.decoder = decoder
self.columns = ColumnsProxy(columns, keyspace, cfam, decoder)
def __get_key(self):
return self.decoder.decode_key(self.keyspace, self.cfam, self._key)
key = property(__get_key)
def __iter__(self):
return iter(self.columns)
def __getitem__(self, idx):
return self.columns[idx]
def __len__(self):
return len(self.columns)
class ColumnsProxy(object):
def __init__(self, columns, keyspace, cfam, decoder):
self.columns = columns
self.keyspace = keyspace
self.cfam = cfam
self.decoder = decoder
def __len__(self):
return len(self.columns)
def __getitem__(self, idx):
return Column(self.decoder.decode_column(self.keyspace,
self.cfam,
self.columns[idx].name,
self.columns[idx].value))
def __iter__(self):
for c in self.columns:
yield Column(self.decoder.decode_column(self.keyspace,
self.cfam,
c.name,
c.value))
def __str__(self):
return "ColumnsProxy(columns=%s)" % self.columns
def __repr__(self):
return str(self)
class Column(object):
def __init__(self, (name, value)):
self.name = name
self.value = value
def __str__(self):
return "Column(%s, %s)" % (self.name, self.value)
def __repr__(self):
return str(self)
if idx == 0 and self._first_vals:
yield self._first_vals
else:
self.description, vals = self.decoder.decode_row(self.ks, self.cf, r)
yield vals
idx += 1

25
drivers/py/cqlsh
View File

@ -26,14 +26,11 @@ import os
import re
try:
from cql import Connection
from cql.errors import CQLException
from cql.results import RowsProxy
import cql
except ImportError:
sys.path.append(os.path.abspath(os.path.dirname(__file__)))
from cql import Connection
from cql.errors import CQLException
from cql.results import RowsProxy
import cql
from cql.results import ResultSet
HISTORY = os.path.join(os.path.expanduser('~'), '.cqlsh')
CQLTYPES = ("bytes", "ascii", "utf8", "timeuuid", "uuid", "long", "int")
@ -55,10 +52,7 @@ class Shell(cmd.Cmd):
def __init__(self, hostname, port, color=False, username=None,
password=None):
cmd.Cmd.__init__(self)
self.conn = Connection(hostname,
port=port,
username=username,
password=password)
self.conn = cql.connect(hostname, port, user=username, password=password)
if os.path.exists(HISTORY):
readline.read_history_file(HISTORY)
@ -92,10 +86,11 @@ class Shell(cmd.Cmd):
statement = self.get_statement(arg)
if not statement: return
result = self.conn.execute(statement)
cursor = self.conn.cursor()
cursor.execute(statement)
if isinstance(result, RowsProxy):
for row in result:
if isinstance(cursor.result, ResultSet):
for row in cursor.result.rows:
self.printout(row.key, BLUE, False)
for column in row.columns:
self.printout(" | ", newline=False)
@ -105,7 +100,7 @@ class Shell(cmd.Cmd):
self.printout(repr(column.value), YELLOW, False)
self.printout("")
else:
if result: print result
if cursor.result: print cursor.result[0]
def emptyline(self):
pass
@ -221,7 +216,7 @@ if __name__ == '__main__':
except SystemExit:
readline.write_history_file(HISTORY)
break
except CQLException, cqlerr:
except cql.Error, cqlerr:
shell.printerr(str(cqlerr), color=RED)
except KeyboardInterrupt:
shell.reset_statement()

8
drivers/py/test/test_query_compression.py
View File

@ -15,18 +15,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from os.path import abspath, exists, join, dirname
if exists(join(abspath(dirname(__file__)), '..', 'cql')):
import sys; sys.path.append(join(abspath(dirname(__file__)), '..'))
import unittest, zlib
from cql import Connection
class TestCompression(unittest.TestCase):
def test_gzip(self):
"compressing a string w/ gzip"
query = "SELECT \"foo\" FROM Standard1 WHERE KEY = \"bar\";"
compressed = Connection.compress_query(query, 'GZIP')
compressed = zlib.compress(query)
decompressed = zlib.decompress(compressed)
assert query == decompressed, "Decompressed query did not match"

23
drivers/py/test/test_query_preparation.py
View File

@ -16,26 +16,26 @@
# limitations under the License.
import unittest
import cql
from cql.marshal import prepare
from cql.errors import InvalidQueryFormat
# TESTS[i] ARGUMENTS[i] -> STANDARDS[i]
TESTS = (
"""
SELECT ?,?,?,? FROM ColumnFamily WHERE KEY = ? AND 'col' = ?;
SELECT :a,:b,:c,:d FROM ColumnFamily WHERE KEY = :e AND 'col' = :f;
""",
"""
USE Keyspace;
""",
"""
SELECT ?..? FROM ColumnFamily;
SELECT :a..:b FROM ColumnFamily;
""",
)
ARGUMENTS = (
(1, 3, long(1000), long(3000), "key", unicode("val")),
tuple(),
("a'b", "c'd'e"),
{'a': 1, 'b': 3, 'c': long(1000), 'd': long(3000), 'e': "key", 'f': unicode("val")},
{},
{'a': "a'b", 'b': "c'd'e"},
)
STANDARDS = (
@ -54,13 +54,12 @@ class TestPrepare(unittest.TestCase):
def test_prepares(self):
"test prepared queries against known standards"
for (i, test) in enumerate(TESTS):
a = prepare(test, *ARGUMENTS[i])
a = prepare(test, **ARGUMENTS[i])
b = STANDARDS[i]
assert a == b, "\n%s !=\n%s" % (a, b)
def test_bad(self):
"ensure bad calls raise exceptions"
self.assertRaises(InvalidQueryFormat, prepare, "? ?", 1)
self.assertRaises(InvalidQueryFormat, prepare, "? ?", 1, 2, 3)
self.assertRaises(InvalidQueryFormat, prepare, "none", 1)
self.assertRaises(KeyError, prepare, ":a :b", a=1)
self.assertRaises(cql.ProgrammingError, prepare, ":a :b", a=1, b=2, c=3)
self.assertRaises(cql.ProgrammingError, prepare, "none", a=1)

0
lib/licenses/guava-r05.txt → lib/licenses/guava-r08.txt
View File

12
src/java/org/apache/cassandra/cache/AutoSavingCache.java vendored
View File

@ -36,6 +36,7 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CompactionManager;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.io.CompactionInfo;
import org.apache.cassandra.io.CompactionType;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.service.StorageService;
@ -203,9 +204,18 @@ public abstract class AutoSavingCache<K, V> extends InstrumentingCache<K, V>
bytes += translateKey(key).remaining();
// an approximation -- the keyset can change while saving
estimatedTotalBytes = bytes;
CompactionType type;
if (cacheType.equals(ColumnFamilyStore.CacheType.KEY_CACHE_TYPE))
type = CompactionType.KEY_CACHE_SAVE;
else if (cacheType.equals(ColumnFamilyStore.CacheType.ROW_CACHE_TYPE))
type = CompactionType.ROW_CACHE_SAVE;
else
type = CompactionType.UNKNOWN;
info = new CompactionInfo(ksname,
cfname,
"Save " + getCachePath().getName(),
type,
0,
estimatedTotalBytes);
}

2
src/java/org/apache/cassandra/cli/Cli.g
View File

@ -119,7 +119,7 @@ package org.apache.cassandra.cli;
if (e instanceof NoViableAltException)
{
errorMessage = "Command not found: `" + this.input + "`. Type 'help' or '?' for help.";
errorMessage = "Command not found: `" + this.input + "`. Type 'help;' or '?' for help.";
}
else
{

101
src/java/org/apache/cassandra/cli/CliClient.java
View File

@ -17,7 +17,9 @@
*/
package org.apache.cassandra.cli;
import java.io.IOError;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
@ -34,6 +36,7 @@ import org.apache.cassandra.db.ColumnFamilyStoreMBean;
import org.apache.cassandra.db.CompactionManagerMBean;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.io.CompactionInfo;
import org.apache.cassandra.io.CompactionType;
import org.apache.cassandra.locator.SimpleSnitch;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.thrift.*;
@ -43,9 +46,13 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.thrift.TBaseHelper;
import org.apache.thrift.TException;
import org.yaml.snakeyaml.constructor.Constructor;
import org.yaml.snakeyaml.Loader;
import org.yaml.snakeyaml.TypeDescription;
import org.yaml.snakeyaml.Yaml;
// Cli Client Side Library
public class CliClient extends CliUserHelp
public class CliClient
{
/**
@ -98,6 +105,34 @@ public class CliClient extends CliUserHelp
STRATEGY_OPTIONS
}
/*
* the <i>add column family</i> command requires a list of arguments,
* this enum defines which arguments are valid.
*/
protected enum ColumnFamilyArgument
{
COLUMN_TYPE,
COMPARATOR,
SUBCOMPARATOR,
COMMENT,
ROWS_CACHED,
ROW_CACHE_SAVE_PERIOD,
KEYS_CACHED,
KEY_CACHE_SAVE_PERIOD,
READ_REPAIR_CHANCE,
GC_GRACE,
COLUMN_METADATA,
MEMTABLE_OPERATIONS,
MEMTABLE_THROUGHPUT,
MEMTABLE_FLUSH_AFTER,
DEFAULT_VALIDATION_CLASS,
MIN_COMPACTION_THRESHOLD,
MAX_COMPACTION_THRESHOLD,
REPLICATE_ON_WRITE,
ROW_CACHE_PROVIDER,
KEY_VALIDATION_CLASS
}
private static final String DEFAULT_PLACEMENT_STRATEGY = "org.apache.cassandra.locator.NetworkTopologyStrategy";
private Cassandra.Client thriftClient = null;
@ -106,13 +141,45 @@ public class CliClient extends CliUserHelp
private String username = null;
private Map<String, KsDef> keyspacesMap = new HashMap<String, KsDef>();
private Map<String, AbstractType> cfKeysComparators;
private ConsistencyLevel consistencyLevel = ConsistencyLevel.ONE;
private ConsistencyLevel consistencyLevel = ConsistencyLevel.ONE;
private CliUserHelp help;
public CliClient(CliSessionState cliSessionState, Cassandra.Client thriftClient)
{
this.sessionState = cliSessionState;
this.thriftClient = thriftClient;
this.cfKeysComparators = new HashMap<String, AbstractType>();
help = getHelp();
}
private CliUserHelp getHelp()
{
final InputStream is = CliClient.class.getClassLoader().getResourceAsStream("org/apache/cassandra/cli/CliHelp.yaml");
assert is != null;
try
{
final Constructor constructor = new Constructor(CliUserHelp.class);
TypeDescription desc = new TypeDescription(CliUserHelp.class);
desc.putListPropertyType("commands", CliCommandHelp.class);
final Yaml yaml = new Yaml(new Loader(constructor));
return (CliUserHelp)yaml.load(is);
}
finally
{
try
{
is.close();
}
catch (IOException e)
{
throw new IOError(e);
}
}
}
public void printBanner()
{
sessionState.out.println(help.banner);
}
// Execute a CLI Statement
@ -134,7 +201,7 @@ public class CliClient extends CliUserHelp
executeGetWithConditions(tree);
break;
case CliParser.NODE_HELP:
printCmdHelp(tree, sessionState);
executeHelp(tree);
break;
case CliParser.NODE_THRIFT_SET:
executeSet(tree);
@ -240,7 +307,27 @@ public class CliClient extends CliUserHelp
return keyspacesMap.get(keyspace);
}
private void executeHelp(Tree tree)
{
if (tree.getChildCount() > 0)
{
String token = tree.getChild(0).getText();
for (CliCommandHelp ch : help.commands)
{
if (token.equals(ch.name))
{
sessionState.out.println(ch.help);
break;
}
}
}
else
{
sessionState.out.println(help.help);
}
}
private void executeCount(Tree statement)
throws TException, InvalidRequestException, UnavailableException, TimedOutException
{
@ -677,7 +764,7 @@ public class CliClient extends CliUserHelp
// table.cf['key']
if (columnSpecCnt == 0)
{
sessionState.err.println("No column name specified, (type 'help' or '?' for help on syntax).");
sessionState.err.println("No column name specified, (type 'help;' or '?' for help on syntax).");
return;
}
// table.cf['key']['column'] = 'value'
@ -1552,7 +1639,7 @@ public class CliClient extends CliUserHelp
for (CompactionInfo info : compactionManagerMBean.getCompactions())
{
// if ongoing compaction type is index build
if (!info.getTaskType().contains("index build"))
if (info.getTaskType() != CompactionType.INDEX_BUILD)
continue;
sessionState.out.printf("%nCurrently building index %s, completed %d of %d bytes.%n",
info.getColumnFamily(),

24
src/java/org/apache/cassandra/cli/CliCommandHelp.java Normal file
View File

@ -0,0 +1,24 @@
/**
* 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.cli;
public class CliCommandHelp
{
public String name;
public String help;
}

8
src/java/org/apache/cassandra/cli/CliMain.java
View File

@ -181,12 +181,6 @@ public class CliMain
}
}
private static void printBanner()
{
sessionState.out.println("Welcome to cassandra CLI.\n");
sessionState.out.println("Type 'help;' or '?' for help. Type 'quit;' or 'exit;' to quit.");
}
/**
* Checks whether the thrift client is connected.
* @return boolean - true when connected, false otherwise
@ -315,7 +309,7 @@ public class CliMain
sessionState.out.close();
}
printBanner();
cliClient.printBanner();
String prompt;
String line = "";

397
src/java/org/apache/cassandra/cli/CliUserHelp.java
View File

@ -17,401 +17,16 @@
*/
package org.apache.cassandra.cli;
import java.util.EnumMap;
import org.antlr.runtime.tree.Tree;
import java.util.List;
/**
* @author Pavel A. Yaskevich
*/
public class CliUserHelp {
public class CliUserHelp
{
public String banner;
/*
* the <i>add column family</i> command requires a list of arguments,
* this enum defines which arguments are valid.
*/
protected enum ColumnFamilyArgument
{
COLUMN_TYPE,
COMPARATOR,
SUBCOMPARATOR,
COMMENT,
ROWS_CACHED,
ROW_CACHE_SAVE_PERIOD,
KEYS_CACHED,
KEY_CACHE_SAVE_PERIOD,
READ_REPAIR_CHANCE,
GC_GRACE,
COLUMN_METADATA,
MEMTABLE_OPERATIONS,
MEMTABLE_THROUGHPUT,
MEMTABLE_FLUSH_AFTER,
DEFAULT_VALIDATION_CLASS,
MIN_COMPACTION_THRESHOLD,
MAX_COMPACTION_THRESHOLD,
REPLICATE_ON_WRITE,
ROW_CACHE_PROVIDER,
KEY_VALIDATION_CLASS
}
protected EnumMap<ColumnFamilyArgument, String> argumentExplanations = new EnumMap<ColumnFamilyArgument, String>(ColumnFamilyArgument.class)
{{
put(ColumnFamilyArgument.COLUMN_TYPE, "Super or Standard");
put(ColumnFamilyArgument.COMMENT, "Human-readable column family description. Any string is acceptable");
put(ColumnFamilyArgument.COMPARATOR, "The class used as a comparator when sorting column names.\n Valid options include: AsciiType, BytesType, LexicalUUIDType,\n LongType, IntegerType, TimeUUIDType, and UTF8Type");
put(ColumnFamilyArgument.SUBCOMPARATOR, "Comparator for sorting subcolumn names, for Super columns only");
put(ColumnFamilyArgument.MEMTABLE_OPERATIONS, "Flush memtables after this many operations (in millions)");
put(ColumnFamilyArgument.MEMTABLE_THROUGHPUT, "... or after this many MB have been written");
put(ColumnFamilyArgument.MEMTABLE_FLUSH_AFTER, "... or after this many minutes");
put(ColumnFamilyArgument.ROWS_CACHED, "Number or percentage of rows to cache");
put(ColumnFamilyArgument.ROW_CACHE_SAVE_PERIOD, "Period with which to persist the row cache, in seconds");
put(ColumnFamilyArgument.KEYS_CACHED, "Number or percentage of keys to cache");
put(ColumnFamilyArgument.KEY_CACHE_SAVE_PERIOD, "Period with which to persist the key cache, in seconds");
put(ColumnFamilyArgument.READ_REPAIR_CHANCE, "Probability (0.0-1.0) with which to perform read repairs on CL.ONE reads");
put(ColumnFamilyArgument.GC_GRACE, "Discard tombstones after this many seconds");
put(ColumnFamilyArgument.MIN_COMPACTION_THRESHOLD, "Avoid minor compactions of less than this number of sstable files");
put(ColumnFamilyArgument.MAX_COMPACTION_THRESHOLD, "Compact no more than this number of sstable files at once");
put(ColumnFamilyArgument.REPLICATE_ON_WRITE, "Replicate every counter update from the leader to the follower replicas");
put(ColumnFamilyArgument.ROW_CACHE_PROVIDER, "Row cache provider, opions are SerializingProvider/ConcurrentLinkedHashCacheProvider");
}};
protected void printCmdHelp(Tree statement, CliSessionState state)
{
if (statement.getChildCount() > 0)
{
int helpType = statement.getChild(0).getType();
switch(helpType)
{
case CliParser.NODE_HELP:
state.out.println("help <command>;\n");
state.out.println("Display the general help page with a list of available commands.");
break;
case CliParser.NODE_CONNECT:
state.out.println("connect <hostname>/<port> (<username> '<password>')?;\n");
state.out.println("Connect to the specified host on the specified port (using specified username and password).\n");
state.out.println("example:");
state.out.println("connect localhost/9160;");
state.out.println("connect localhost/9160 user 'badpasswd';");
state.out.println("connect 127.0.0.1/9160 user 'badpasswd';");
break;
case CliParser.NODE_USE_TABLE:
state.out.println("use <keyspace>;");
state.out.println("use <keyspace> <username> '<password>';\n");
state.out.println("Switch to the specified keyspace. The optional username and password fields");
state.out.println("are needed when performing authentication.\n");
break;
case CliParser.NODE_DESCRIBE_TABLE:
state.out.println("describe keyspace (<keyspace>)?;\n");
state.out.println("Show additional information about the specified keyspace.");
state.out.println("Command could be used without <keyspace> argument if you are already authenticated to keyspace.\n");
state.out.println("example:");
state.out.println("describe keyspace system;");
break;
case CliParser.NODE_DESCRIBE_CLUSTER:
state.out.println("describe cluster;\n");
state.out.println("Display information about cluster: snitch, partitioner, schema versions.");
break;
case CliParser.NODE_EXIT:
state.out.println("exit;");
state.out.println("quit;\n");
state.out.println("Exit this utility.");
break;
case CliParser.NODE_SHOW_CLUSTER_NAME:
state.out.println("show cluster name;\n");
state.out.println("Displays the name of the currently connected cluster.");
break;
case CliParser.NODE_SHOW_VERSION:
state.out.println("show api version;\n");
state.out.println("Displays the API version number.");
break;
case CliParser.NODE_SHOW_KEYSPACES:
state.out.println("show keyspaces;\n");
state.out.println("Displays a list of the keyspaces available on the currently connected cluster.");
break;
case CliParser.NODE_ADD_KEYSPACE:
state.out.println("create keyspace <keyspace>;");
state.out.println("create keyspace <keyspace> with <att1>=<value1>;");
state.out.println("create keyspace <keyspace> with <att1>=<value1> and <att2>=<value2> ...;\n");
state.out.println("Create a new keyspace with the specified values for the given set of attributes.\n");
state.out.println("valid attributes are:");
state.out.println(" placement_strategy: the fully qualified class used to place replicas in");
state.out.println(" this keyspace. Valid values are");
state.out.println(" org.apache.cassandra.locator.SimpleStrategy,");
state.out.println(" org.apache.cassandra.locator.NetworkTopologyStrategy,");
state.out.println(" and org.apache.cassandra.locator.OldNetworkTopologyStrategy");
state.out.println(" strategy_options: additional options for placement_strategy.");
state.out.println(" Simple and OldNetworkTopology strategies require");
state.out.println(" 'replication_factor':N, and NetworkTopologyStrategy");
state.out.println(" requires a map of 'DCName':N per-DC.");
state.out.println("\nexamples:");
state.out.println("create keyspace foo with");
state.out.println(" placement_strategy = 'org.apache.cassandra.locator.SimpleStrategy'");
state.out.println(" and strategy_options=[{replication_factor:3}];");
state.out.println("create keyspace foo with");
state.out.println(" placement_strategy = 'org.apache.cassandra.locator.NetworkTopologyStrategy';");
state.out.println(" and strategy_options=[{DC1:2, DC2:2}];");
break;
case CliParser.NODE_UPDATE_KEYSPACE:
state.out.println("update keyspace <keyspace>;");
state.out.println("update keyspace <keyspace> with <att1>=<value1>;");
state.out.println("update keyspace <keyspace> with <att1>=<value1> and <att2>=<value2> ...;\n");
state.out.println("Update a keyspace with the specified values for the given set of attributes.\n");
state.out.println("valid attributes are:");
state.out.println(" placement_strategy: the fully qualified class used to place replicas in");
state.out.println(" this keyspace. Valid values are");
state.out.println(" org.apache.cassandra.locator.SimpleStrategy,");
state.out.println(" org.apache.cassandra.locator.NetworkTopologyStrategy,");
state.out.println(" and org.apache.cassandra.locator.OldNetworkTopologyStrategy");
state.out.println(" strategy_options: additional options for placement_strategy.");
state.out.println(" Simple and OldNetworkTopology strategies require");
state.out.println(" 'replication_factor':N, and NetworkTopologyStrategy");
state.out.println(" requires a map of 'DCName':N per-DC.");
state.out.println("\nexamples:");
state.out.println("update keyspace foo with");
state.out.println(" placement_strategy = 'org.apache.cassandra.locator.SimpleStrategy'");
state.out.println(" and strategy_options=[{replication_factor:4}];");
state.out.println("update keyspace foo with");
state.out.println(" placement_strategy = 'org.apache.cassandra.locator.NetworkTopologyStrategy';");
state.out.println(" and strategy_options=[{DC1:3, DC2:2}];");
break;
case CliParser.NODE_ADD_COLUMN_FAMILY:
state.out.println("create column family Bar;");
state.out.println("create column family Bar with <att1>=<value1>;");
state.out.println("create column family Bar with <att1>=<value1> and <att2>=<value2>...;\n");
state.out.println("Create a new column family with the specified values for the given set of");
state.out.println("attributes. Note that you must be using a keyspace.\n");
state.out.println("valid attributes are:");
for (ColumnFamilyArgument argument : ColumnFamilyArgument.values())
state.out.printf(" - %s: %s%n", argument.toString().toLowerCase(), argumentExplanations.get(argument));
state.out.println(" - column_metadata: Metadata which describes columns of column family.");
state.out.println(" Supported format is [{ k:v, k:v, ... }, { ... }, ...]");
state.out.println(" Valid attributes: column_name, validation_class (see comparator),");
state.out.println(" index_type (integer), index_name.");
state.out.println("example:\n");
state.out.println("create column family Bar with column_type = 'Super' and comparator = 'AsciiType'");
state.out.println(" and rows_cached = 10000;");
state.out.println("create column family Baz with comparator = 'LongType' and rows_cached = 10000;");
state.out.print("create column family Foo with comparator=LongType and column_metadata=");
state.out.print("[{ column_name:Test, validation_class:IntegerType, index_type:0, index_name:IdxName");
state.out.println("}, { column_name:'other name', validation_class:LongType }];");
break;
case CliParser.NODE_UPDATE_COLUMN_FAMILY:
state.out.println("update column family Bar;");
state.out.println("update column family Bar with <att1>=<value1>;");
state.out.println("update column family Bar with <att1>=<value1> and <att2>=<value2>...;\n");
state.out.println("Update a column family with the specified values for the given set of");
state.out.println("attributes. Note that you must be using a keyspace.\n");
state.out.println("valid attributes are:");
for (ColumnFamilyArgument argument : ColumnFamilyArgument.values())
{
if (argument == ColumnFamilyArgument.COMPARATOR || argument == ColumnFamilyArgument.SUBCOMPARATOR)
continue;
state.out.printf(" - %s: %s%n", argument.toString().toLowerCase(), argumentExplanations.get(argument));
}
state.out.println(" - column_metadata: Metadata which describes columns of column family.");
state.out.println(" Supported format is [{ k:v, k:v, ... }, { ... }, ...]");
state.out.println(" Valid attributes: column_name, validation_class (see comparator),");
state.out.println(" index_type (integer), index_name.");
state.out.println("example:\n");
state.out.print("update column family Foo with column_metadata=");
state.out.print("[{ column_name:Test, validation_class:IntegerType, index_type:0, index_name:IdxName");
state.out.println("}] and rows_cached=100 and comment='this is helpful comment.';");
break;
case CliParser.NODE_DEL_KEYSPACE:
state.out.println("drop keyspace <keyspace>;\n");
state.out.println("Drops the specified keyspace.\n");
state.out.println("example:");
state.out.println("drop keyspace foo;");
break;
case CliParser.NODE_DEL_COLUMN_FAMILY:
state.out.println("drop column family <name>;\n");
state.out.println("Drops the specified column family.\n");
state.out.println("example:");
state.out.println("drop column family foo;");
break;
case CliParser.NODE_THRIFT_GET :
state.out.println("get <cf>['<key>'];");
state.out.println("get <cf>['<key>']['<col>'] (as <type>)*;");
state.out.println("get <cf>['<key>']['<super>'];");
state.out.println("get <cf>['<key>'][<function>];");
state.out.println("get <cf>['<key>'][<function>(<super>)][<function>(<col>)];");
state.out.println("get <cf> where <column> = <value> [and <column> > <value> and ...] [limit <integer>];");
state.out.println("Default LIMIT is 100. Available operations: =, >, >=, <, <=\n");
state.out.println("get <cf>['<key>']['<super>']['<col>'] (as <type>)*;");
state.out.print("Note: `as <type>` is optional, it dynamically converts column value to the specified type");
state.out.println(", column value validator will be set to <type>.");
state.out.println("Available functions: " + CliClient.Function.getFunctionNames());
state.out.println("Available types: IntegerType, LongType, UTF8Type, ASCIIType, TimeUUIDType, LexicalUUIDType.\n");
state.out.println("examples:");
state.out.println("get bar[testkey];");
state.out.println("get bar[testkey][test_column] as IntegerType;");
state.out.println("get bar[testkey][utf8(hello)];");
break;
case CliParser.NODE_THRIFT_SET:
state.out.println("set <cf>['<key>']['<col>'] = <value>;");
state.out.println("set <cf>['<key>']['<super>']['<col>'] = <value>;");
state.out.println("set <cf>['<key>']['<col>'] = <function>(<argument>);");
state.out.println("set <cf>['<key>']['<super>']['<col>'] = <function>(<argument>);");
state.out.println("set <cf>[<key>][<function>(<col>)] = <value> || <function>;");
state.out.println("set <cf>[<key>][<function>(<col>) || <col>] = <value> || <function> with ttl = <secs>;");
state.out.println("Available functions: " + CliClient.Function.getFunctionNames() + "\n");
state.out.println("examples:");
state.out.println("set bar['testkey']['my super']['test col']='this is a test';");
state.out.println("set baz['testkey']['test col']='this is also a test';");
state.out.println("set diz[testkey][testcol] = utf8('this is utf8 string.');");
state.out.println("set bar[testkey][timeuuid()] = utf('hello world');");
state.out.println("set bar[testkey][timeuuid()] = utf('hello world') with ttl = 30;");
state.out.println("set diz[testkey][testcol] = 'this is utf8 string.' with ttl = 150;");
break;
case CliParser.NODE_THRIFT_INCR:
state.out.println("incr <cf>['<key>']['<col>'] [by <value>];");
state.out.println("incr <cf>['<key>']['<super>']['<col>'] [by <value>];");
state.out.println("examples:");
state.out.println("incr bar['testkey']['my super']['test col'];");
state.out.println("incr bar['testkey']['my super']['test col'] by 42;");
state.out.println("incr baz['testkey']['test col'] by -4;");
break;
case CliParser.NODE_THRIFT_DECR:
state.out.println("decr <cf>['<key>']['<col>'] [by <value>];");
state.out.println("decr <cf>['<key>']['<super>']['<col>'] [by <value>];");
state.out.println("examples:");
state.out.println("decr bar['testkey']['my super']['test col'];");
state.out.println("decr bar['testkey']['my super']['test col'] by 42;");
state.out.println("decr baz['testkey']['test col'] by 10;");
break;
case CliParser.NODE_THRIFT_DEL:
state.out.println("del <cf>['<key>'];");
state.out.println("del <cf>['<key>']['<col>'];");
state.out.println("del <cf>['<key>']['<super>']['<col>'];\n");
state.out.println("Deletes a record, a column, or a subcolumn.\n");
state.out.println("example:");
state.out.println("del bar['testkey']['my super']['test col'];");
state.out.println("del baz['testkey']['test col'];");
state.out.println("del baz['testkey'];");
break;
case CliParser.NODE_THRIFT_COUNT:
state.out.println("count <cf>['<key>'];");
state.out.println("count <cf>['<key>']['<super>'];\n");
state.out.println("Count the number of columns in the specified key or subcolumns in the specified");
state.out.println("super column.\n");
state.out.println("example:");
state.out.println("count bar['testkey']['my super'];");
state.out.println("count baz['testkey'];");
break;
case CliParser.NODE_LIST:
state.out.println("list <cf>;");
state.out.println("list <cf>[<startKey>:];");
state.out.println("list <cf>[<startKey>:<endKey>];");
state.out.println("list ... limit N;");
state.out.println("List a range of rows in the column or supercolumn family.\n");
state.out.println("example:");
state.out.println("list Users[j:] limit 40;");
break;
case CliParser.NODE_TRUNCATE:
state.out.println("truncate <column_family>;");
state.out.println("Truncate specified column family.\n");
state.out.println("example:");
state.out.println("truncate Category;");
break;
case CliParser.NODE_ASSUME:
state.out.println("assume <column_family> comparator as <type>;");
state.out.println("assume <column_family> sub_comparator as <type>;");
state.out.println("assume <column_family> validator as <type>;");
state.out.println("assume <column_family> keys as <type>;\n");
state.out.println("Assume one of the attributes (comparator, sub_comparator, validator or keys)");
state.out.println("of the given column family to match specified type. Available types: " + CliClient.Function.getFunctionNames());
state.out.println("example:");
state.out.println("assume Users comparator as lexicaluuid;");
break;
case CliParser.NODE_CONSISTENCY_LEVEL:
state.out.println("consistencylevel as <level>");
state.out.println("example:");
state.out.println("consistencylevel as QUORUM");
break;
default:
state.out.println("?");
break;
}
}
else
{
state.out.println("List of all CLI commands:");
state.out.println("? Display this message.");
state.out.println("help; Display this help.");
state.out.println("help <command>; Display detailed, command-specific help.");
state.out.println("connect <hostname>/<port> (<username> '<password>')?; Connect to thrift service.");
state.out.println("use <keyspace> [<username> 'password']; Switch to a keyspace.");
state.out.println("describe keyspace (<keyspacename>)?; Describe keyspace.");
state.out.println("exit; Exit CLI.");
state.out.println("quit; Exit CLI.");
state.out.println("describe cluster; Display information about cluster.");
state.out.println("show cluster name; Display cluster name.");
state.out.println("show keyspaces; Show list of keyspaces.");
state.out.println("show api version; Show server API version.");
state.out.println("create keyspace <keyspace> [with <att1>=<value1> [and <att2>=<value2> ...]];");
state.out.println(" Add a new keyspace with the specified attribute(s) and value(s).");
state.out.println("update keyspace <keyspace> [with <att1>=<value1> [and <att2>=<value2> ...]];");
state.out.println(" Update a keyspace with the specified attribute(s) and value(s).");
state.out.println("create column family <cf> [with <att1>=<value1> [and <att2>=<value2> ...]];");
state.out.println(" Create a new column family with the specified attribute(s) and value(s).");
state.out.println("update column family <cf> [with <att1>=<value1> [and <att2>=<value2> ...]];");
state.out.println(" Update a column family with the specified attribute(s) and value(s).");
state.out.println("drop keyspace <keyspace>; Delete a keyspace.");
state.out.println("drop column family <cf>; Delete a column family.");
state.out.println("get <cf>['<key>']; Get a slice of columns.");
state.out.println("get <cf>['<key>']['<super>']; Get a slice of sub columns.");
state.out.println("get <cf> where <column> = <value> [and <column> > <value> and ...] [limit int]; ");
state.out.println("get <cf>['<key>']['<col>'] (as <type>)*; Get a column value.");
state.out.println("get <cf>['<key>']['<super>']['<col>'] (as <type>)*; Get a sub column value.");
state.out.println("set <cf>['<key>']['<col>'] = <value> (with ttl = <secs>)*; Set a column.");
state.out.println("set <cf>['<key>']['<super>']['<col>'] = <value> (with ttl = <secs>)*;");
state.out.println(" Set a sub column.");
state.out.println("del <cf>['<key>']; Delete record.");
state.out.println("del <cf>['<key>']['<col>']; Delete column.");
state.out.println("del <cf>['<key>']['<super>']['<col>']; Delete sub column.");
state.out.println("count <cf>['<key>']; Count columns in record.");
state.out.println("count <cf>['<key>']['<super>']; Count columns in a super column.");
state.out.println("incr <cf>['<key>']['<col>'] [by <value>]; Increment a counter column.");
state.out.println("incr <cf>['<key>']['<super>']['<col>'] [by <value>];");
state.out.println(" Increment a counter sub-column.");
state.out.println("decr <cf>['<key>']['<col>'] [by <value>]; Decrement a counter column.");
state.out.println("decr <cf>['<key>']['<super>']['<col>'] [by <value>];");
state.out.println(" Decrement a counter sub-column.");
state.out.println("truncate <column_family>; Truncate specified column family.");
state.out.println("assume <column_family> <attribute> as <type>;");
state.out.println(" Assume a given column family attributes to match a specified type.");
state.out.println("consistencylevel as <level>;");
state.out.println(" Change the consistency level for set,get, and list operations.");
state.out.println("list <cf>; List all rows in the column family.");
state.out.println("list <cf>[<startKey>:];");
state.out.println(" List rows in the column family beginning with <startKey>.");
state.out.println("list <cf>[<startKey>:<endKey>];");
state.out.println(" List rows in the column family in the range from <startKey> to <endKey>.");
state.out.println("list ... limit N; Limit the list results to N.");
}
}
public String help;
public List<CliCommandHelp> commands;
}

1
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -657,6 +657,7 @@ public final class CFMetaData
if (cf_def.isSetMerge_shards_chance()) { newCFMD.mergeShardsChance(cf_def.merge_shards_chance); }
if (cf_def.isSetRow_cache_provider()) { newCFMD.rowCacheProvider(FBUtilities.newCacheProvider(cf_def.row_cache_provider)); }
if (cf_def.isSetKey_alias()) { newCFMD.keyAlias(cf_def.key_alias); }
if (cf_def.isSetKey_validation_class()) { newCFMD.keyValidator(DatabaseDescriptor.getComparator(cf_def.key_validation_class)); }
return newCFMD.comment(cf_def.comment)
.rowCacheSize(cf_def.row_cache_size)

11
src/java/org/apache/cassandra/db/CompactionManager.java
View File

@ -28,7 +28,6 @@ import java.security.MessageDigest;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.management.MBeanServer;
@ -529,7 +528,9 @@ public class CompactionManager implements CompactionManagerMBean
// so in our single-threaded compaction world this is a valid way of determining if we're compacting
// all the sstables (that existed when we started)
boolean major = cfs.isCompleteSSTables(sstables);
String type = major ? "Major" : "Minor";
CompactionType type = major
? CompactionType.MAJOR
: CompactionType.MINOR;
logger.info("Compacting {}: {}", type, sstables);
long startTime = System.currentTimeMillis();
@ -1147,7 +1148,7 @@ public class CompactionManager implements CompactionManagerMBean
{
public ValidationCompactionIterator(ColumnFamilyStore cfs, Range range) throws IOException
{
super("Validation",
super(CompactionType.VALIDATION,
getCollatingIterator(cfs.getSSTables(), range),
new CompactionController(cfs, cfs.getSSTables(), true, getDefaultGcBefore(cfs), false));
}
@ -1348,7 +1349,7 @@ public class CompactionManager implements CompactionManagerMBean
{
return new CompactionInfo(sstable.descriptor.ksname,
sstable.descriptor.cfname,
"Cleanup of " + sstable.getColumnFamilyName(),
CompactionType.CLEANUP,
scanner.getFilePointer(),
scanner.getFileLength());
}
@ -1375,7 +1376,7 @@ public class CompactionManager implements CompactionManagerMBean
{
return new CompactionInfo(sstable.descriptor.ksname,
sstable.descriptor.cfname,
"Scrub " + sstable,
CompactionType.SCRUB,
dataFile.getFilePointer(),
dataFile.length());
}

3
src/java/org/apache/cassandra/db/Table.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.dht.LocalToken;
import org.apache.cassandra.io.CompactionInfo;
import org.apache.cassandra.io.CompactionType;
import org.apache.cassandra.io.sstable.ReducingKeyIterator;
import org.apache.cassandra.io.sstable.SSTableDeletingReference;
import org.apache.cassandra.io.sstable.SSTableReader;
@ -614,7 +615,7 @@ public class Table
{
return new CompactionInfo(cfs.table.name,
cfs.columnFamily,
String.format("Secondary index build %s", cfs.columnFamily),
CompactionType.INDEX_BUILD,
iter.getTotalBytes(),
iter.getBytesRead());
}

8
src/java/org/apache/cassandra/io/CompactionInfo.java
View File

@ -23,13 +23,15 @@ import java.io.Serializable;
/** Implements serializable to allow structured info to be returned via JMX. */
public final class CompactionInfo implements Serializable
{
private final String ksname;
private final String cfname;
private final String tasktype;
private final CompactionType tasktype;
private final long bytesComplete;
private final long totalBytes;
public CompactionInfo(String ksname, String cfname, String tasktype, long bytesComplete, long totalBytes)
public CompactionInfo(String ksname, String cfname, CompactionType tasktype, long bytesComplete, long totalBytes)
{
this.ksname = ksname;
this.cfname = cfname;
@ -64,7 +66,7 @@ public final class CompactionInfo implements Serializable
return totalBytes;
}
public String getTaskType()
public CompactionType getTaskType()
{
return tasktype;
}

10
src/java/org/apache/cassandra/io/CompactionIterator.java
View File

@ -24,18 +24,14 @@ package org.apache.cassandra.io;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import org.apache.commons.collections.iterators.CollatingIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CompactionManager;
import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
import org.apache.cassandra.io.sstable.SSTableReader;
@ -53,7 +49,7 @@ implements Closeable, CompactionInfo.Holder
public static final int FILE_BUFFER_SIZE = 1024 * 1024;
protected final List<SSTableIdentityIterator> rows = new ArrayList<SSTableIdentityIterator>();
protected final String type;
protected final CompactionType type;
protected final CompactionController controller;
private long totalBytes;
@ -68,13 +64,13 @@ implements Closeable, CompactionInfo.Holder
// current target bytes to compact per millisecond
private int targetBytesPerMS = -1;
public CompactionIterator(String type, Iterable<SSTableReader> sstables, CompactionController controller) throws IOException
public CompactionIterator(CompactionType type, Iterable<SSTableReader> sstables, CompactionController controller) throws IOException
{
this(type, getCollatingIterator(sstables), controller);
}
@SuppressWarnings("unchecked")
protected CompactionIterator(String type, Iterator iter, CompactionController controller)
protected CompactionIterator(CompactionType type, Iterator iter, CompactionController controller)
{
super(iter);
this.type = type;

27
src/java/org/apache/cassandra/io/CompactionType.java Normal file
View File

@ -0,0 +1,27 @@
package org.apache.cassandra.io;
public enum CompactionType
{
MAJOR("Major"),
MINOR("Minor"),
VALIDATION("Validation"),
KEY_CACHE_SAVE("Key cache save"),
ROW_CACHE_SAVE("Row cache save"),
CLEANUP("Cleanup"),
SCRUB("Scrub"),
INDEX_BUILD("Secondary index build"),
SSTABLE_BUILD("SSTable build"),
UNKNOWN("Unkown compaction type");
private final String type;
CompactionType(String type)
{
this.type = type;
}
public String toString()
{
return type;
}
}

9
src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
View File

@ -28,6 +28,7 @@ import java.util.Set;
import com.google.common.collect.Sets;
import org.apache.cassandra.io.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -38,13 +39,7 @@ import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.marshal.AbstractCommutativeType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.AbstractCompactedRow;
import org.apache.cassandra.io.CompactionController;
import org.apache.cassandra.io.CompactionInfo;
import org.apache.cassandra.io.LazilyCompactedRow;
import org.apache.cassandra.io.PrecompactedRow;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.FileMark;
import org.apache.cassandra.io.util.FileUtils;
@ -271,7 +266,7 @@ public class SSTableWriter extends SSTable
// both file offsets are still valid post-close
return new CompactionInfo(desc.ksname,
desc.cfname,
"SSTable rebuild",
CompactionType.SSTABLE_BUILD,
indexer.dfile.getFilePointer(),
indexer.dfile.length());
}

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

@ -76,6 +76,7 @@ public class StorageProxy implements StorageProxyMBean
private static final WritePerformer standardWritePerformer;
private static final WritePerformer counterWritePerformer;
private static final WritePerformer counterWriteOnCoordinatorPerformer;
public static final StorageProxy instance = new StorageProxy();
@ -102,11 +103,25 @@ public class StorageProxy implements StorageProxyMBean
}
};
/*
* We execute counter writes in 2 places: either directly in the coordinator node if it is a replica, or
* in CounterMutationVerbHandler on a replica othewise. The write must be executed on the MUTATION stage
* but on the latter case, the verb handler already run on the MUTATION stage, so we must not execute the
* underlying on the stage otherwise we risk a deadlock. Hence two different performer.
*/
counterWritePerformer = new WritePerformer()
{
public void apply(IMutation mutation, Multimap<InetAddress, InetAddress> hintedEndpoints, IWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException
{
applyCounterMutation(mutation, hintedEndpoints, responseHandler, localDataCenter, consistency_level);
applyCounterMutation(mutation, hintedEndpoints, responseHandler, localDataCenter, consistency_level, false);
}
};
counterWriteOnCoordinatorPerformer = new WritePerformer()
{
public void apply(IMutation mutation, Multimap<InetAddress, InetAddress> hintedEndpoints, IWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException
{
applyCounterMutation(mutation, hintedEndpoints, responseHandler, localDataCenter, consistency_level, true);
}
};
}
@ -367,7 +382,7 @@ public class StorageProxy implements StorageProxyMBean
if (endpoint.equals(FBUtilities.getLocalAddress()))
{
applyCounterMutationOnLeader(cm);
applyCounterMutationOnCoordinator(cm);
}
else
{
@ -423,7 +438,14 @@ public class StorageProxy implements StorageProxyMBean
write(Collections.singletonList(cm), cm.consistency(), counterWritePerformer, false);
}
private static void applyCounterMutation(final IMutation mutation, final Multimap<InetAddress, InetAddress> hintedEndpoints, final IWriteResponseHandler responseHandler, final String localDataCenter, final ConsistencyLevel consistency_level)
// Same as applyCounterMutationOnLeader but must with the difference that it use the MUTATION stage to execute the write (while
// applyCounterMutationOnLeader assumes it is on the MUTATION stage already)
public static void applyCounterMutationOnCoordinator(CounterMutation cm) throws UnavailableException, TimeoutException, IOException
{
write(Collections.singletonList(cm), cm.consistency(), counterWriteOnCoordinatorPerformer, false);
}
private static void applyCounterMutation(final IMutation mutation, final Multimap<InetAddress, InetAddress> hintedEndpoints, final IWriteResponseHandler responseHandler, final String localDataCenter, final ConsistencyLevel consistency_level, boolean executeOnMutationStage)
{
// we apply locally first, then send it to other replica
if (logger.isDebugEnabled())
@ -456,7 +478,10 @@ public class StorageProxy implements StorageProxyMBean
}
}
};
StageManager.getStage(Stage.MUTATION).execute(runnable);
if (executeOnMutationStage)
StageManager.getStage(Stage.MUTATION).execute(runnable);
else
runnable.run();
}
/**

1150
src/resources/org/apache/cassandra/cli/CliHelp.yaml Normal file
View File

File diff suppressed because it is too large Load Diff

31
test/long/org/apache/cassandra/db/LongCompactionSpeedTest.java
View File

@ -20,28 +20,29 @@ package org.apache.cassandra.db;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.Util;
import org.junit.Test;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableUtils;
import org.apache.cassandra.io.sstable.SSTableWriter;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.streaming.OperationType;
import org.apache.cassandra.utils.ByteBufferUtil;
import java.nio.ByteBuffer;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.NodeId;
import static org.apache.cassandra.db.context.CounterContext.ContextState;
import static junit.framework.Assert.assertEquals;
public class LongCompactionSpeedTest extends CleanupHelper
{
public static final String TABLE1 = "Keyspace1";
public static final InetAddress LOCAL = FBUtilities.getLocalAddress();
/**
* Test compaction with a very wide row.
@ -188,12 +189,12 @@ public class LongCompactionSpeedTest extends CleanupHelper
protected CounterColumn createCounterColumn(String name)
{
byte[] context = Util.concatByteArrays(
FBUtilities.getLocalAddress().getAddress(), FBUtilities.toByteArray(9L), FBUtilities.toByteArray(3L),
FBUtilities.toByteArray(2), FBUtilities.toByteArray(4L), FBUtilities.toByteArray(2L),
FBUtilities.toByteArray(4), FBUtilities.toByteArray(3L), FBUtilities.toByteArray(3L),
FBUtilities.toByteArray(8), FBUtilities.toByteArray(2L), FBUtilities.toByteArray(4L)
);
return new CounterColumn(ByteBufferUtil.bytes(name), ByteBuffer.wrap(context), 0L);
ContextState context = ContextState.allocate(4, 1);
context.writeElement(NodeId.fromInt(1), 4L, 2L, true);
context.writeElement(NodeId.fromInt(2), 4L, 2L);
context.writeElement(NodeId.fromInt(4), 3L, 3L);
context.writeElement(NodeId.fromInt(8), 2L, 4L);
return new CounterColumn(ByteBufferUtil.bytes(name), context.context, 0L);
}
}

664
test/system/test_cql.py
View File

@ -23,18 +23,18 @@ import sys, uuid, time
sys.path.append(join(abspath(dirname(__file__)), '../../drivers/py'))
from cql import Connection
from cql.errors import CQLException
import cql
from cql.connection import Connection
from __init__ import ThriftTester
from __init__ import thrift_client # TODO: temporary
def assert_raises(exception, method, *args):
def assert_raises(exception, method, *args, **kwargs):
try:
method(*args)
method(*args, **kwargs)
except exception:
return
raise AssertionError("failed to see expected exception")
def uuid1bytes_to_millis(uuidbytes):
return (uuid.UUID(bytes=uuidbytes).get_time() / 10000) - 12219292800000L
@ -72,12 +72,12 @@ def load_sample(dbconn):
WITH comparator = ascii AND default_validation = ascii;
""")
dbconn.execute("CREATE INDEX ON IndexedA (birthdate)")
query = "UPDATE StandardString1 SET ? = ?, ? = ? WHERE KEY = ?"
dbconn.execute(query, "ca1", "va1", "col", "val", "ka")
dbconn.execute(query, "cb1", "vb1", "col", "val", "kb")
dbconn.execute(query, "cc1", "vc1", "col", "val", "kc")
dbconn.execute(query, "cd1", "vd1", "col", "val", "kd")
query = "UPDATE StandardString1 SET :c1 = :v1, :c2 = :v2 WHERE KEY = :key"
dbconn.execute(query, dict(c1="ca1", v1="va1", c2="col", v2="val", key="ka"))
dbconn.execute(query, dict(c1="cb1", v1="vb1", c2="col", v2="val", key="kb"))
dbconn.execute(query, dict(c1="cc1", v1="vc1", c2="col", v2="val", key="kc"))
dbconn.execute(query, dict(c1="cd1", v1="vd1", c2="col", v2="val", key="kd"))
dbconn.execute("""
BEGIN BATCH USING CONSISTENCY ONE
@ -90,7 +90,7 @@ def load_sample(dbconn):
UPDATE StandardLongA SET 5='5', 6='6', 7='8', 9='9' WHERE KEY='ag'
APPLY BATCH
""")
dbconn.execute("""
BEGIN BATCH USING CONSISTENCY ONE
UPDATE StandardIntegerA SET 10='a', 20='b', 30='c', 40='d' WHERE KEY='k1';
@ -114,255 +114,294 @@ def load_sample(dbconn):
""")
def init(keyspace="Keyspace1"):
dbconn = Connection('localhost', 9170, keyspace)
load_sample(dbconn)
return dbconn
dbconn = cql.connect('localhost', 9170, keyspace)
cursor = dbconn.cursor()
load_sample(cursor)
return cursor
class TestCql(ThriftTester):
def test_select_simple(self):
"retrieve a column"
conn = init()
r = conn.execute("SELECT 'ca1' FROM StandardString1 WHERE KEY='ka'")
assert r[0].key == 'ka'
assert r[0].columns[0].name == 'ca1'
assert r[0].columns[0].value == 'va1'
cursor = init()
cursor.execute("SELECT 'ca1' FROM StandardString1 WHERE KEY='ka'")
r = cursor.fetchone()
d = cursor.description
assert d[0][0] == cql.ROW_KEY
assert r[0] == 'ka'
assert d[1][0] == 'ca1'
assert r[1] == 'va1'
def test_select_columns(self):
"retrieve multiple columns"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
assert "cd1" in [i.name for i in r[0].columns]
assert "col" in [i.name for i in r[0].columns]
d = cursor.description
assert "cd1" in [col_dscptn[0] for col_dscptn in d]
assert "col" in [col_dscptn[0] for col_dscptn in d]
def test_select_row_range(self):
"retrieve a range of rows with columns"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT 4 FROM StandardLongA WHERE KEY > 'ad' AND KEY < 'ag';
""")
assert len(r) == 4
assert r[0].key == "ad"
assert r[1].key == "ae"
assert r[2].key == "af"
assert len(r[0].columns) == 1
assert len(r[1].columns) == 1
assert len(r[2].columns) == 1
keys = ['ad', 'ae', 'af']
assert cursor.rowcount == 4
for i in range(3):
r = cursor.fetchone()
assert len(r) == 2
assert r[0] == keys[i]
def test_select_row_range_with_limit(self):
"retrieve a limited range of rows with columns"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT 1,5,9 FROM StandardLongA WHERE KEY > 'aa'
AND KEY < 'ag' LIMIT 3
""")
assert len(r) == 3
r = conn.execute("""
assert cursor.rowcount == 3
cursor.execute("""
SELECT 20,40 FROM StandardIntegerA WHERE KEY > 'k1'
AND KEY < 'k7' LIMIT 5
""")
assert len(r) == 5
r[0].key == "k1"
r[4].key == "k5"
assert cursor.rowcount == 5
for i in range(5):
r = cursor.fetchone()
assert r[0] == "k%d" % (i+1)
def test_select_columns_slice(self):
"range of columns (slice) by row"
conn = init()
r = conn.execute("SELECT 1..3 FROM StandardLongA WHERE KEY = 'aa';")
assert len(r) == 1
assert r[0].columns[0].value == "1"
assert r[0].columns[1].value == "2"
assert r[0].columns[2].value == "3"
r = conn.execute("SELECT 10..30 FROM StandardIntegerA WHERE KEY='k1'")
assert len(r) == 1
assert r[0].columns[0].value == "a"
assert r[0].columns[1].value == "b"
assert r[0].columns[2].value == "c"
cursor = init()
cursor.execute("SELECT 1..3 FROM StandardLongA WHERE KEY = 'aa';")
assert cursor.rowcount == 1
r = cursor.fetchone()
assert r[0] == "aa"
assert r[1] == "1"
assert r[2] == "2"
assert r[3] == "3"
cursor.execute("SELECT 10..30 FROM StandardIntegerA WHERE KEY='k1'")
assert cursor.rowcount == 1
r = cursor.fetchone()
assert r[0] == "k1"
assert r[1] == "a"
assert r[2] == "b"
assert r[3] == "c"
def test_select_columns_slice_all(self):
"slice all columns in a row"
conn = init()
r = conn.execute("SELECT * FROM StandardString1 WHERE KEY = 'ka';")
assert len(r[0].columns) == 2
r = conn.execute("SELECT ''..'' FROM StandardString1 WHERE KEY = 'ka';")
assert len(r[0].columns) == 2
cursor = init()
cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'ka';")
r = cursor.fetchone()
assert len(r) == 3
cursor.execute("SELECT ''..'' FROM StandardString1 WHERE KEY = 'ka';")
r = cursor.fetchone()
assert len(r) == 3
def test_select_columns_slice_with_limit(self):
"range of columns (slice) by row with limit"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT FIRST 1 1..3 FROM StandardLongA WHERE KEY = 'aa';
""")
assert len(r) == 1
assert len(r[0].columns) == 1
assert r[0].columns[0].value == "1"
assert cursor.rowcount == 1
r = cursor.fetchone()
assert len(r) == 2
assert r[0] == "aa"
assert r[1] == "1"
def test_select_columns_slice_reversed(self):
"range of columns (slice) by row reversed"
conn = init()
r = conn.execute("""
cursor= init()
cursor.execute("""
SELECT FIRST 2 REVERSED 3..1 FROM StandardLongA WHERE KEY = 'aa';
""")
assert len(r) == 1, "%d != 1" % len(r)
assert len(r[0].columns) == 2
assert r[0].columns[0].value == "3"
assert r[0].columns[1].value == "2"
assert cursor.rowcount == 1, "%d != 1" % cursor.rowcount
r = cursor.fetchone()
assert len(r) == 3
assert r[0] == 'aa'
assert r[1] == "3"
assert r[2] == "2"
def test_error_on_multiple_key_by(self):
"ensure multiple key-bys in where clause excepts"
conn = init()
assert_raises(CQLException, conn.execute, """
cursor = init()
assert_raises(cql.ProgrammingError, cursor.execute, """
SELECT 'col' FROM StandardString1 WHERE KEY = 'ka' AND KEY = 'kb';
""")
def test_index_scan_equality(self):
"indexed scan where column equals value"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
""")
assert cursor.rowcount == 2
r = cursor.fetchone()
assert r[0] == "asmith"
assert len(r) == 2
r = cursor.fetchone()
assert r[0] == "dozer"
assert len(r) == 2
assert r[0].key == "asmith"
assert r[1].key == "dozer"
assert len(r[0].columns) == 1
assert len(r[1].columns) == 1
def test_index_scan_greater_than(self):
"indexed scan where a column is greater than a value"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
AND 'unindexed' > 200
""")
assert len(r) == 1
assert r[0].key == "asmith"
assert cursor.rowcount == 1
r = cursor.fetchone()
assert r[0] == "asmith"
def test_index_scan_with_start_key(self):
"indexed scan with a starting key"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
AND KEY > 'asmithZ'
""")
assert len(r) == 1
assert r[0].key == "dozer"
assert cursor.rowcount == 1
r = cursor.fetchone()
assert r[0] == "dozer"
def test_no_where_clause(self):
"empty where clause (range query w/o start key)"
conn = init()
r = conn.execute("SELECT 'col' FROM StandardString1 LIMIT 3")
assert len(r) == 3
assert r[0].key == "ka"
assert r[1].key == "kb"
assert r[2].key == "kc"
cursor = init()
cursor.execute("SELECT 'col' FROM StandardString1 LIMIT 3")
assert cursor.rowcount == 3
rows = cursor.fetchmany(3)
assert rows[0][0] == "ka"
assert rows[1][0] == "kb"
assert rows[2][0] == "kc"
def test_column_count(self):
"getting a result count instead of results"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT COUNT(1..4) FROM StandardLongA WHERE KEY = 'aa';
""")
assert r == 4, "expected 4 results, got %d" % (r and r or 0)
r = cursor.fetchone()
assert r[0] == 4, "expected 4 results, got %d" % (r and r or 0)
def test_truncate_columnfamily(self):
"truncating a column family"
conn = init()
conn.execute('TRUNCATE StandardString1;')
r = conn.execute("SELECT 'cd1' FROM StandardString1 WHERE KEY = 'kd'")
assert len(r) == 0
cursor = init()
cursor.execute('TRUNCATE StandardString1;')
cursor.execute("SELECT 'cd1' FROM StandardString1 WHERE KEY = 'kd'")
assert cursor.rowcount == 0
def test_delete_columns(self):
"delete columns from a row"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
assert "cd1" in [i.name for i in r[0].columns]
assert "col" in [i.name for i in r[0].columns]
conn.execute("""
colnames = [col_d[0] for col_d in cursor.description]
assert "cd1" in colnames
assert "col" in colnames
cursor.execute("""
DELETE 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
r = conn.execute("""
cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
assert len(r[0].columns) == 0
r = cursor.fetchone()
assert len(r) == 1
def test_delete_columns_multi_rows(self):
"delete columns from multiple rows"
conn = init()
r = conn.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
assert len(r[0].columns) == 1
r = conn.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
assert len(r[0].columns) == 1
cursor = init()
conn.execute("""
cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
r = cursor.fetchone()
assert len(r) == 2
cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
r = cursor.fetchone()
assert len(r) == 2
cursor.execute("""
DELETE 'col' FROM StandardString1 WHERE KEY IN ('kc', 'kd')
""")
r = conn.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
assert len(r[0].columns) == 0
r = conn.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
assert len(r[0].columns) == 0
cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
r = cursor.fetchone()
assert len(r) == 1
cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
r = cursor.fetchone()
assert len(r) == 1
def test_delete_rows(self):
"delete entire rows"
conn = init()
r = conn.execute("""
cursor = init()
cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
assert "cd1" in [i.name for i in r[0].columns]
assert "col" in [i.name for i in r[0].columns]
conn.execute("DELETE FROM StandardString1 WHERE KEY = 'kd'")
r = conn.execute("""
colnames = [col_d[0] for col_d in cursor.description]
assert "cd1" in colnames
assert "col" in colnames
cursor.execute("DELETE FROM StandardString1 WHERE KEY = 'kd'")
cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
assert len(r[0].columns) == 0
r = cursor.fetchone()
assert len(r) == 1
def test_create_keyspace(self):
"create a new keyspace"
init().execute("""
cursor = init()
cursor.execute("""
CREATE KEYSPACE TestKeyspace42 WITH strategy_options:DC1 = '1'
AND strategy_class = 'NetworkTopologyStrategy'
""")
# TODO: temporary (until this can be done with CQL).
ksdef = thrift_client.describe_keyspace("TestKeyspace42")
strategy_class = "org.apache.cassandra.locator.NetworkTopologyStrategy"
assert ksdef.strategy_class == strategy_class
assert ksdef.strategy_options['DC1'] == "1"
def test_drop_keyspace(self):
"removing a keyspace"
conn = init()
conn.execute("""
CREATE KEYSPACE Keyspace4Drop
WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1
cursor = init()
cursor.execute("""
CREATE KEYSPACE Keyspace4Drop WITH strategy_options:replication_factor = '1'
AND strategy_class = 'SimpleStrategy'
""")
# TODO: temporary (until this can be done with CQL).
thrift_client.describe_keyspace("Keyspace4Drop")
conn.execute('DROP KEYSPACE Keyspace4Drop;')
cursor.execute('DROP KEYSPACE Keyspace4Drop;')
# Technically this should throw a ttypes.NotFound(), but this is
# temporary and so not worth requiring it on PYTHONPATH.
assert_raises(Exception,
thrift_client.describe_keyspace,
"Keyspace4Drop")
def test_create_column_family(self):
"create a new column family"
conn = init()
conn.execute("""
CREATE KEYSPACE CreateCFKeyspace WITH strategy_options:replication_factor = 1
AND strategy_class = 'SimpleStrategy';
cursor = init()
cursor.execute("""
CREATE KEYSPACE CreateCFKeyspace WITH strategy_options:replication_factor = '1'
AND strategy_class = 'SimpleStrategy';
""")
conn.execute("USE CreateCFKeyspace;")
conn.execute("""
cursor.execute("USE CreateCFKeyspace;")
cursor.execute("""
CREATE COLUMNFAMILY NewCf1 (
KEY varint PRIMARY KEY,
'username' text,
@ -371,7 +410,7 @@ class TestCql(ThriftTester):
'id' uuid
) WITH comment = 'shiny, new, cf' AND default_validation = ascii;
""")
# TODO: temporary (until this can be done with CQL).
ksdef = thrift_client.describe_keyspace("CreateCFKeyspace")
assert len(ksdef.cf_defs) == 1, \
@ -383,27 +422,27 @@ class TestCql(ThriftTester):
assert cfam.default_validation_class == "org.apache.cassandra.db.marshal.AsciiType"
assert cfam.comparator_type == "org.apache.cassandra.db.marshal.UTF8Type"
assert cfam.key_validation_class == "org.apache.cassandra.db.marshal.IntegerType"
# Missing primary key
assert_raises(CQLException, conn.execute, "CREATE COLUMNFAMILY NewCf2")
assert_raises(cql.ProgrammingError, cursor.execute, "CREATE COLUMNFAMILY NewCf2")
# Too many primary keys
assert_raises(CQLException,
conn.execute,
assert_raises(cql.ProgrammingError,
cursor.execute,
"""CREATE COLUMNFAMILY NewCf2
(KEY varint PRIMARY KEY, KEY text PRIMARY KEY)""")
# No column defs
conn.execute("""CREATE COLUMNFAMILY NewCf3
cursor.execute("""CREATE COLUMNFAMILY NewCf3
(KEY varint PRIMARY KEY) WITH comparator = bigint""")
ksdef = thrift_client.describe_keyspace("CreateCFKeyspace")
assert len(ksdef.cf_defs) == 2, \
"expected 3 column families total, found %d" % len(ksdef.cf_defs)
cfam = [i for i in ksdef.cf_defs if i.name == "NewCf3"][0]
assert cfam.comparator_type == "org.apache.cassandra.db.marshal.LongType"
# Column defs, defaults otherwise
conn.execute("""CREATE COLUMNFAMILY NewCf4
cursor.execute("""CREATE COLUMNFAMILY NewCf4
(KEY varint PRIMARY KEY, 'a' varint, 'b' varint)
WITH comparator = text;""")
ksdef = thrift_client.describe_keyspace("CreateCFKeyspace")
@ -415,34 +454,34 @@ class TestCql(ThriftTester):
for coldef in cfam.column_metadata:
assert coldef.name in ("a", "b"), "Unknown column name " + coldef.name
assert coldef.validation_class.endswith("marshal.IntegerType")
def test_drop_columnfamily(self):
"removing a column family"
conn = init()
conn.execute("""
CREATE KEYSPACE Keyspace4CFDrop WITH strategy_options:replication_factor = 1
AND strategy_class = 'SimpleStrategy';
cursor = init()
cursor.execute("""
CREATE KEYSPACE Keyspace4CFDrop WITH strategy_options:replication_factor = '1'
AND strategy_class = 'SimpleStrategy';
""")
conn.execute('USE Keyspace4CFDrop;')
conn.execute('CREATE COLUMNFAMILY CF4Drop (KEY varint PRIMARY KEY);')
cursor.execute('USE Keyspace4CFDrop;')
cursor.execute('CREATE COLUMNFAMILY CF4Drop (KEY varint PRIMARY KEY);')
# TODO: temporary (until this can be done with CQL).
ksdef = thrift_client.describe_keyspace("Keyspace4CFDrop")
assert len(ksdef.cf_defs), "Column family not created!"
conn.execute('DROP COLUMNFAMILY CF4Drop;')
cursor.execute('DROP COLUMNFAMILY CF4Drop;')
ksdef = thrift_client.describe_keyspace("Keyspace4CFDrop")
assert not len(ksdef.cf_defs), "Column family not deleted!"
def test_create_indexs(self):
"creating column indexes"
conn = init()
conn.execute("USE Keyspace1")
conn.execute("CREATE COLUMNFAMILY CreateIndex1 (KEY text PRIMARY KEY)")
conn.execute("CREATE INDEX namedIndex ON CreateIndex1 (items)")
conn.execute("CREATE INDEX ON CreateIndex1 (stuff)")
cursor = init()
cursor.execute("USE Keyspace1")
cursor.execute("CREATE COLUMNFAMILY CreateIndex1 (KEY text PRIMARY KEY)")
cursor.execute("CREATE INDEX namedIndex ON CreateIndex1 (items)")
cursor.execute("CREATE INDEX ON CreateIndex1 (stuff)")
# TODO: temporary (until this can be done with CQL).
ksdef = thrift_client.describe_keyspace("Keyspace1")
cfam = [i for i in ksdef.cf_defs if i.name == "CreateIndex1"][0]
@ -455,205 +494,208 @@ class TestCql(ThriftTester):
assert stuff.index_type == 0, "missing index"
# already indexed
assert_raises(CQLException,
conn.execute,
assert_raises(cql.ProgrammingError,
cursor.execute,
"CREATE INDEX ON CreateIndex1 (stuff)")
def test_time_uuid(self):
"store and retrieve time-based (type 1) uuids"
conn = init()
cursor = init()
# Store and retrieve a timeuuid using it's hex-formatted string
timeuuid = uuid.uuid1()
conn.execute("""
cursor.execute("""
UPDATE StandardTimeUUID SET '%s' = 10 WHERE KEY = 'uuidtest'
""" % str(timeuuid))
r = conn.execute("""
cursor.execute("""
SELECT '%s' FROM StandardTimeUUID WHERE KEY = 'uuidtest'
""" % str(timeuuid))
assert r[0].columns[0].name == timeuuid
d = cursor.description
assert d[1][0] == timeuuid, "%s, %s" % (str(d[1][0]), str(timeuuid))
# Tests a node-side conversion from bigint to UUID.
ms = uuid1bytes_to_millis(uuid.uuid1().bytes)
conn.execute("""
cursor.execute("""
UPDATE StandardTimeUUIDValues SET 'id' = %d WHERE KEY = 'uuidtest'
""" % ms)
r = conn.execute("""
cursor.execute("""
SELECT 'id' FROM StandardTimeUUIDValues WHERE KEY = 'uuidtest'
""")
assert uuid1bytes_to_millis(r[0].columns[0].value.bytes) == ms
r = cursor.fetchone()
assert uuid1bytes_to_millis(r[1].bytes) == ms
# Tests a node-side conversion from ISO8601 to UUID.
conn.execute("""
cursor.execute("""
UPDATE StandardTimeUUIDValues SET 'id2' = '2011-01-31 17:00:00-0000'
WHERE KEY = 'uuidtest'
""")
r = conn.execute("""
cursor.execute("""
SELECT 'id2' FROM StandardTimeUUIDValues WHERE KEY = 'uuidtest'
""")
# 2011-01-31 17:00:00-0000 == 1296493200000ms
ms = uuid1bytes_to_millis(r[0].columns[0].value.bytes)
r = cursor.fetchone()
ms = uuid1bytes_to_millis(r[1].bytes)
assert ms == 1296493200000, \
"%d != 1296493200000 (2011-01-31 17:00:00-0000)" % ms
# Tests node-side conversion of timeuuid("now") to UUID
conn.execute("""
cursor.execute("""
UPDATE StandardTimeUUIDValues SET 'id3' = 'now'
WHERE KEY = 'uuidtest'
""")
r = conn.execute("""
cursor.execute("""
SELECT 'id3' FROM StandardTimeUUIDValues WHERE KEY = 'uuidtest'
""")
ms = uuid1bytes_to_millis(r[0].columns[0].value.bytes)
r = cursor.fetchone()
ms = uuid1bytes_to_millis(r[1].bytes)
assert ((time.time() * 1e3) - ms) < 100, \
"new timeuuid not within 100ms of now (UPDATE vs. SELECT)"
uuid_range = []
update = "UPDATE StandardTimeUUID SET ? = ? WHERE KEY = slicetest"
update = "UPDATE StandardTimeUUID SET :name = :val WHERE KEY = slicetest"
for i in range(5):
uuid_range.append(uuid.uuid1())
conn.execute(update, uuid_range[i], i)
cursor.execute(update, dict(name=uuid_range[i], val=i))
cursor.execute("""
SELECT :start..:finish FROM StandardTimeUUID WHERE KEY = slicetest
""", dict(start=uuid_range[0], finish=uuid_range[len(uuid_range)-1]))
d = cursor.description
for (i, col_d) in enumerate(d[1:]):
assert uuid_range[i] == col_d[0]
r = conn.execute("""
SELECT ?..? FROM StandardTimeUUID WHERE KEY = slicetest
""", uuid_range[0], uuid_range[len(uuid_range)-1])
for (i, col) in enumerate(r[0]):
assert uuid_range[i] == col.name
def test_lexical_uuid(self):
"store and retrieve lexical uuids"
conn = init()
cursor = init()
uid = uuid.uuid4()
conn.execute("UPDATE StandardUUID SET ? = 10 WHERE KEY = 'uuidtest'",
uid)
r = conn.execute("SELECT ? FROM StandardUUID WHERE KEY = 'uuidtest'",
uid)
assert r[0].columns[0].name == uid, r[0].columns[0].name
cursor.execute("UPDATE StandardUUID SET :name = 10 WHERE KEY = 'uuidtest'",
dict(name=uid))
cursor.execute("SELECT :name FROM StandardUUID WHERE KEY = 'uuidtest'",
dict(name=uid))
d = cursor.description
assert d[1][0] == uid, "expected %s, got %s (%s)" % \
(uid.bytes.encode('hex'), str(d[1][0]).encode('hex'), d[1][1])
# TODO: slices of uuids from cf w/ LexicalUUIDType comparator
def test_utf8_read_write(self):
"reading and writing utf8 values"
conn = init()
cursor = init()
# Sorting: ¢ (u00a2) < © (u00a9) < ® (u00ae) < ¿ (u00bf)
conn.execute("UPDATE StandardUtf82 SET ? = v1 WHERE KEY = k1", "¿")
conn.execute("UPDATE StandardUtf82 SET ? = v1 WHERE KEY = k1", "©")
conn.execute("UPDATE StandardUtf82 SET ? = v1 WHERE KEY = k1", "®")
conn.execute("UPDATE StandardUtf82 SET ? = v1 WHERE KEY = k1", "¢")
r = conn.execute("SELECT * FROM StandardUtf82 WHERE KEY = k1")
assert r[0][0].name == u"¢"
assert r[0][1].name == u"©"
assert r[0][2].name == u"®"
assert r[0][3].name == u"¿"
r = conn.execute("SELECT ?..'' FROM StandardUtf82 WHERE KEY = k1", "©")
assert len(r[0]) == 3
assert r[0][0].name == u"©"
assert r[0][1].name == u"®"
assert r[0][2].name == u"¿"
cursor.execute("UPDATE StandardUtf82 SET :name = v1 WHERE KEY = k1", dict(name="¿"))
cursor.execute("UPDATE StandardUtf82 SET :name = v1 WHERE KEY = k1", dict(name="©"))
cursor.execute("UPDATE StandardUtf82 SET :name = v1 WHERE KEY = k1", dict(name="®"))
cursor.execute("UPDATE StandardUtf82 SET :name = v1 WHERE KEY = k1", dict(name="¢"))
cursor.execute("SELECT * FROM StandardUtf82 WHERE KEY = k1")
d = cursor.description
assert d[1][0] == u"¢", d[1][0]
assert d[2][0] == u"©", d[2][0]
assert d[3][0] == u"®", d[3][0]
assert d[4][0] == u"¿", d[4][0]
cursor.execute("SELECT :start..'' FROM StandardUtf82 WHERE KEY = k1", dict(start="©"))
r = cursor.fetchone()
assert len(r) == 4
d = cursor.description
assert d[1][0] == u"©"
assert d[2][0] == u"®"
assert d[3][0] == u"¿"
def test_read_write_negative_numerics(self):
"reading and writing negative numeric values"
conn = init()
cursor = init()
for cf in ("StandardIntegerA", "StandardLongA"):
for i in range(10):
conn.execute("UPDATE ? SET ? = ? WHERE KEY = negatives;",
cf,
-(i + 1),
i)
r = conn.execute("SELECT ?..? FROM ? WHERE KEY = negatives;",
-10,
-1,
cf)
assert len(r[0]) == 10, \
"returned %d columns, expected %d" % (len(r[0]), 10)
assert r[0][0].name == -10
assert r[0][9].name == -1
cursor.execute("UPDATE :cf SET :name = :val WHERE KEY = negatives;",
dict(cf=cf, name=-(i + 1), val=i))
cursor.execute("SELECT :start..:finish FROM :cf WHERE KEY = negatives;",
dict(start=-10, finish=-1, cf=cf))
r = cursor.fetchone()
assert len(r) == 11, \
"returned %d columns, expected %d" % (len(r) - 1, 10)
d = cursor.description
assert d[1][0] == -10
assert d[10][0] == -1
def test_escaped_quotes(self):
"reading and writing strings w/ escaped quotes"
conn = init()
conn.execute("""
UPDATE StandardString1 SET 'x''and''y' = z WHERE KEY = ?
""", "test_escaped_quotes")
r = conn.execute("""
SELECT 'x''and''y' FROM StandardString1 WHERE KEY = ?
""", "test_escaped_quotes")
assert (len(r) == 1) and (len(r[0]) == 1), "wrong number of results"
assert r[0][0].name == "x\'and\'y"
cursor = init()
cursor.execute("""
UPDATE StandardString1 SET 'x''and''y' = z WHERE KEY = :key
""", dict(key="test_escaped_quotes"))
cursor.execute("""
SELECT 'x''and''y' FROM StandardString1 WHERE KEY = :key
""", dict(key="test_escaped_quotes"))
assert cursor.rowcount == 1
r = cursor.fetchone()
assert len(r) == 2, "wrong number of results"
d = cursor.description
assert d[1][0] == "x\'and\'y"
def test_typed_keys(self):
"using typed keys"
conn = init()
r = conn.execute("SELECT * FROM StandardString1 WHERE KEY = ?", "ka")
assert isinstance(r[0].key, unicode), \
"wrong key-type returned, expected unicode, got %s" % type(r[0].key)
cursor = init()
cursor.execute("SELECT * FROM StandardString1 WHERE KEY = :key", dict(key="ka"))
r = cursor.fetchone()
assert isinstance(r[0], unicode), \
"wrong key-type returned, expected unicode, got %s" % type(r[0])
# FIXME: The above is woefully inadequate, but the test config uses
# CollatingOrderPreservingPartitioner which only supports UTF8.
def test_write_using_insert(self):
"peforming writes using \"insert\""
conn = init()
conn.execute("INSERT INTO StandardUtf82 (KEY, ?, ?) VALUES (?, ?, ?)",
"pork",
"beef",
"meat",
"bacon",
"brisket")
r = conn.execute("SELECT * FROM StandardUtf82 WHERE KEY = ?", "meat")
assert r[0][0].name == "beef"
assert r[0][0].value == "brisket"
assert r[0][1].name == "pork"
assert r[0][1].value == "bacon"
cursor = init()
cursor.execute("INSERT INTO StandardUtf82 (KEY, :c1, :c2) VALUES (:key, :v1, :v2)",
dict(c1="pork", c2="beef", key="meat", v1="bacon", v2="brisket"))
cursor.execute("SELECT * FROM StandardUtf82 WHERE KEY = :key", dict(key="meat"))
r = cursor.fetchone()
d = cursor.description
assert d[1][0] == "beef"
assert r[1] == "brisket"
assert d[2][0] == "pork"
assert r[2] == "bacon"
# Bad writes.
# Too many column values
assert_raises(CQLException,
conn.execute,
"INSERT INTO StandardUtf82 (KEY, ?) VALUES (?, ?, ?)",
"name1",
"key0",
"value1",
"value2")
assert_raises(cql.ProgrammingError,
cursor.execute,
"INSERT INTO StandardUtf82 (KEY, :c1) VALUES (:key, :v1, :v2)",
dict(c1="name1", key="key0", v1="value1", v2="value2"))
# Too many column names, (not enough column values)
assert_raises(CQLException,
conn.execute,
"INSERT INTO StandardUtf82 (KEY, ?, ?) VALUES (?, ?)",
"name1",
"name2",
"key0",
"value1")
assert_raises(cql.ProgrammingError,
cursor.execute,
"INSERT INTO StandardUtf82 (KEY, :c1, :c2) VALUES (:key, :v1)",
dict(c1="name1", c2="name2", key="key0", v1="value1"))
def test_compression_disabled(self):
"reading and writing w/ compression disabled"
conn = init()
conn.execute("UPDATE StandardString1 SET ? = ? WHERE KEY = ?",
"some_name",
"some_value",
"compression_test",
compression='NONE')
r = conn.execute("SELECT ? FROM StandardString1 WHERE KEY = ?",
"some_name",
"compression_test",
compression='NONE')
assert len(r) == 1, "expected 1 result, got %d" % len(r)
assert r[0][0].name == "some_name", \
"unrecognized name '%s'" % r[0][0].name
assert r[0][0].value == "some_value", \
"unrecognized value '%s'" % r[0][0].value
cursor = init()
cursor.compression = 'NONE'
cursor.execute("UPDATE StandardString1 SET :name = :val WHERE KEY = :key",
dict(name="some_name", val="some_value", key="compression_test"))
cursor.execute("SELECT :name FROM StandardString1 WHERE KEY = :key",
dict(name="some_name", key="compression_test"))
assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount
colnames = [col_d[0] for col_d in cursor.description]
assert colnames[1] == "some_name", \
"unrecognized name '%s'" % colnames[1]
r = cursor.fetchone()
assert r[1] == "some_value", \
"unrecognized value '%s'" % r[1]

6
test/unit/org/apache/cassandra/io/LazilyCompactedRowTest.java
View File

@ -27,9 +27,7 @@ import java.io.*;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HashSet;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import org.apache.cassandra.CleanupHelper;
@ -309,7 +307,7 @@ public class LazilyCompactedRowTest extends CleanupHelper
{
public LazyCompactionIterator(Iterable<SSTableReader> sstables, CompactionController controller) throws IOException
{
super("Lazy", sstables, controller);
super(CompactionType.UNKNOWN, sstables, controller);
}
@Override
@ -323,7 +321,7 @@ public class LazilyCompactedRowTest extends CleanupHelper
{
public PreCompactingIterator(Iterable<SSTableReader> sstables, CompactionController controller) throws IOException
{
super("Pre", sstables, controller);
super(CompactionType.UNKNOWN, sstables, controller);
}
@Override

13
tools/stress/src/org/apache/cassandra/stress/Session.java
View File

@ -229,9 +229,6 @@ public class Session
if (cmd.hasOption("g"))
keysPerCall = Integer.parseInt(cmd.getOptionValue("g"));
if (cmd.hasOption("l"))
replicationStrategyOptions.put("replication_factor", String.valueOf(Integer.parseInt(cmd.getOptionValue("l"))));
if (cmd.hasOption("e"))
consistencyLevel = ConsistencyLevel.valueOf(cmd.getOptionValue("e").toUpperCase());
@ -241,6 +238,11 @@ public class Session
if (cmd.hasOption("R"))
replicationStrategy = cmd.getOptionValue("R");
if (cmd.hasOption("l"))
replicationStrategyOptions.put("replication_factor", String.valueOf(Integer.parseInt(cmd.getOptionValue("l"))));
else if (replicationStrategy.endsWith("SimpleStrategy"))
replicationStrategyOptions.put("replication_factor", "1");
if (cmd.hasOption("O"))
{
String[] pairs = StringUtils.split(cmd.getOptionValue("O"), ',');
@ -421,19 +423,18 @@ public class Session
keyspace.setCf_defs(new ArrayList<CfDef>(Arrays.asList(standardCfDef, superCfDef, counterCfDef, counterSuperCfDef)));
Cassandra.Client client = getClient(false);
try
{
client.system_add_keyspace(keyspace);
out.println(String.format("Created keyspaces. Sleeping %ss for propagation.", nodes.length));
Thread.sleep(nodes.length * 1000); // seconds
}
catch (InvalidRequestException e)
{
out.println(e.getWhy());
out.println("Unable to create stress keyspace: " + e.getWhy());
System.exit(1);
}
catch (Exception e)
{