diff --git a/CHANGES.txt b/CHANGES.txt
index 8462f706dc..52a6284b34 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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
diff --git a/build.xml b/build.xml
index 622a1576e9..506eedabb9 100644
--- a/build.xml
+++ b/build.xml
@@ -27,6 +27,7 @@
+
@@ -383,7 +384,9 @@
-
+
+
+
diff --git a/drivers/py/README b/drivers/py/README
index d783d22ad7..8beb9f06f1 100644
--- a/drivers/py/README
+++ b/drivers/py/README
@@ -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")
diff --git a/drivers/py/cql/__init__.py b/drivers/py/cql/__init__.py
index 4790696086..52d8702cf0 100644
--- a/drivers/py/cql/__init__.py
+++ b/drivers/py/cql/__init__.py
@@ -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
diff --git a/drivers/py/cql/connection.py b/drivers/py/cql/connection.py
index 754f360f75..9e7bc4df17 100644
--- a/drivers/py/cql/connection.py
+++ b/drivers/py/cql/connection.py
@@ -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)
diff --git a/drivers/py/cql/cursor.py b/drivers/py/cql/cursor.py
new file mode 100644
index 0000000000..c83e8afc38
--- /dev/null
+++ b/drivers/py/cql/cursor.py
@@ -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, ))
diff --git a/drivers/py/cql/decoders.py b/drivers/py/cql/decoders.py
index 651fa98045..20c065e9d9 100644
--- a/drivers/py/cql/decoders.py
+++ b/drivers/py/cql/decoders.py
@@ -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
diff --git a/drivers/py/cql/errors.py b/drivers/py/cql/errors.py
index 8021616c5a..fa87cd26c5 100644
--- a/drivers/py/cql/errors.py
+++ b/drivers/py/cql/errors.py
@@ -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
diff --git a/drivers/py/cql/marshal.py b/drivers/py/cql/marshal.py
index 641d56ffbd..c8fac8a1ec 100644
--- a/drivers/py/cql/marshal.py
+++ b/drivers/py/cql/marshal.py
@@ -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"(? 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:
diff --git a/drivers/py/cql/results.py b/drivers/py/cql/results.py
index c878516148..68b27f8b70 100644
--- a/drivers/py/cql/results.py
+++ b/drivers/py/cql/results.py
@@ -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
diff --git a/drivers/py/cqlsh b/drivers/py/cqlsh
index a99b9ba5a9..9b701c0b1c 100755
--- a/drivers/py/cqlsh
+++ b/drivers/py/cqlsh
@@ -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()
diff --git a/drivers/py/test/test_query_compression.py b/drivers/py/test/test_query_compression.py
index a0aff7155f..7d161c7a11 100644
--- a/drivers/py/test/test_query_compression.py
+++ b/drivers/py/test/test_query_compression.py
@@ -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"
diff --git a/drivers/py/test/test_query_preparation.py b/drivers/py/test/test_query_preparation.py
index 5cf202907f..95d39a8fcc 100644
--- a/drivers/py/test/test_query_preparation.py
+++ b/drivers/py/test/test_query_preparation.py
@@ -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)
diff --git a/lib/licenses/guava-r05.txt b/lib/licenses/guava-r08.txt
similarity index 100%
rename from lib/licenses/guava-r05.txt
rename to lib/licenses/guava-r08.txt
diff --git a/src/java/org/apache/cassandra/cache/AutoSavingCache.java b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
index 566d4c09e2..7609672246 100644
--- a/src/java/org/apache/cassandra/cache/AutoSavingCache.java
+++ b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
@@ -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 extends InstrumentingCache
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);
}
diff --git a/src/java/org/apache/cassandra/cli/Cli.g b/src/java/org/apache/cassandra/cli/Cli.g
index e043378f7a..1b77fee997 100644
--- a/src/java/org/apache/cassandra/cli/Cli.g
+++ b/src/java/org/apache/cassandra/cli/Cli.g
@@ -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
{
diff --git a/src/java/org/apache/cassandra/cli/CliClient.java b/src/java/org/apache/cassandra/cli/CliClient.java
index 44d063056d..45b110b43b 100644
--- a/src/java/org/apache/cassandra/cli/CliClient.java
+++ b/src/java/org/apache/cassandra/cli/CliClient.java
@@ -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 add column family 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 keyspacesMap = new HashMap();
private Map 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();
+ 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(),
diff --git a/src/java/org/apache/cassandra/cli/CliCommandHelp.java b/src/java/org/apache/cassandra/cli/CliCommandHelp.java
new file mode 100644
index 0000000000..d4e3b34c07
--- /dev/null
+++ b/src/java/org/apache/cassandra/cli/CliCommandHelp.java
@@ -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;
+}
\ No newline at end of file
diff --git a/src/java/org/apache/cassandra/cli/CliMain.java b/src/java/org/apache/cassandra/cli/CliMain.java
index 4331895e47..47541d684a 100644
--- a/src/java/org/apache/cassandra/cli/CliMain.java
+++ b/src/java/org/apache/cassandra/cli/CliMain.java
@@ -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 = "";
diff --git a/src/java/org/apache/cassandra/cli/CliUserHelp.java b/src/java/org/apache/cassandra/cli/CliUserHelp.java
index 5dee2462f1..a64bf8791e 100644
--- a/src/java/org/apache/cassandra/cli/CliUserHelp.java
+++ b/src/java/org/apache/cassandra/cli/CliUserHelp.java
@@ -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 add column family 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 argumentExplanations = new EnumMap(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 ;\n");
- state.out.println("Display the general help page with a list of available commands.");
- break;
- case CliParser.NODE_CONNECT:
- state.out.println("connect / ( '')?;\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 ;");
- state.out.println("use '';\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 ()?;\n");
- state.out.println("Show additional information about the specified keyspace.");
- state.out.println("Command could be used without 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 ;");
- state.out.println("create keyspace with =;");
- state.out.println("create keyspace with = and = ...;\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 ;");
- state.out.println("update keyspace with =;");
- state.out.println("update keyspace with = and = ...;\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 =;");
- state.out.println("create column family Bar with = and =...;\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 =;");
- state.out.println("update column family Bar with = and =...;\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 ;\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 ;\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 [''];");
- state.out.println("get [''][''] (as )*;");
- state.out.println("get [''][''];");
- state.out.println("get [''][];");
- state.out.println("get [''][()][()];");
- state.out.println("get where = [and > and ...] [limit ];");
- state.out.println("Default LIMIT is 100. Available operations: =, >, >=, <, <=\n");
- state.out.println("get [''][''][''] (as )*;");
- state.out.print("Note: `as ` is optional, it dynamically converts column value to the specified type");
- state.out.println(", column value validator will be set to .");
- 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 [''][''] = ;");
- state.out.println("set [''][''][''] = ;");
- state.out.println("set [''][''] = ();");
- state.out.println("set [''][''][''] = ();");
- state.out.println("set [][()] = || ;");
- state.out.println("set [][() || ] = || with ttl = ;");
- 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 [''][''] [by ];");
- state.out.println("incr [''][''][''] [by ];");
- 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 [''][''] [by ];");
- state.out.println("decr [''][''][''] [by ];");
- 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 [''];");
- state.out.println("del [''][''];");
- state.out.println("del [''][''][''];\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 [''];");
- state.out.println("count [''][''];\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 ;");
- state.out.println("list [:];");
- state.out.println("list [:];");
- 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 ;");
- 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 comparator as ;");
- state.out.println("assume sub_comparator as ;");
- state.out.println("assume validator as ;");
- state.out.println("assume keys as ;\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 ");
- 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 ; Display detailed, command-specific help.");
- state.out.println("connect / ( '')?; Connect to thrift service.");
- state.out.println("use [ 'password']; Switch to a keyspace.");
- state.out.println("describe keyspace ()?; 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 [with = [and = ...]];");
- state.out.println(" Add a new keyspace with the specified attribute(s) and value(s).");
- state.out.println("update keyspace [with = [and = ...]];");
- state.out.println(" Update a keyspace with the specified attribute(s) and value(s).");
- state.out.println("create column family [with = [and = ...]];");
- state.out.println(" Create a new column family with the specified attribute(s) and value(s).");
- state.out.println("update column family [with = [and = ...]];");
- state.out.println(" Update a column family with the specified attribute(s) and value(s).");
- state.out.println("drop keyspace ; Delete a keyspace.");
- state.out.println("drop column family ; Delete a column family.");
- state.out.println("get ['']; Get a slice of columns.");
- state.out.println("get ['']['']; Get a slice of sub columns.");
- state.out.println("get where = [and > and ...] [limit int]; ");
- state.out.println("get [''][''] (as )*; Get a column value.");
- state.out.println("get [''][''][''] (as )*; Get a sub column value.");
- state.out.println("set [''][''] = (with ttl = )*; Set a column.");
- state.out.println("set [''][''][''] = (with ttl = )*;");
- state.out.println(" Set a sub column.");
- state.out.println("del ['']; Delete record.");
- state.out.println("del ['']['']; Delete column.");
- state.out.println("del ['']['']['']; Delete sub column.");
- state.out.println("count ['']; Count columns in record.");
- state.out.println("count ['']['']; Count columns in a super column.");
- state.out.println("incr [''][''] [by ]; Increment a counter column.");
- state.out.println("incr [''][''][''] [by ];");
- state.out.println(" Increment a counter sub-column.");
- state.out.println("decr [''][''] [by ]; Decrement a counter column.");
- state.out.println("decr [''][''][''] [by ];");
- state.out.println(" Decrement a counter sub-column.");
- state.out.println("truncate ; Truncate specified column family.");
- state.out.println("assume as ;");
- state.out.println(" Assume a given column family attributes to match a specified type.");
- state.out.println("consistencylevel as ;");
- state.out.println(" Change the consistency level for set,get, and list operations.");
- state.out.println("list ; List all rows in the column family.");
- state.out.println("list [:];");
- state.out.println(" List rows in the column family beginning with .");
- state.out.println("list [:];");
- state.out.println(" List rows in the column family in the range from to .");
- state.out.println("list ... limit N; Limit the list results to N.");
- }
- }
+ public String help;
+ public List commands;
}
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index c6b1e18ca8..bff93d9e76 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -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)
diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java
index 8369d2efc0..1ecbbcad5b 100644
--- a/src/java/org/apache/cassandra/db/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/CompactionManager.java
@@ -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());
}
diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java
index 4161882a0d..77c385623d 100644
--- a/src/java/org/apache/cassandra/db/Table.java
+++ b/src/java/org/apache/cassandra/db/Table.java
@@ -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());
}
diff --git a/src/java/org/apache/cassandra/io/CompactionInfo.java b/src/java/org/apache/cassandra/io/CompactionInfo.java
index f4d1d29515..147197d89d 100644
--- a/src/java/org/apache/cassandra/io/CompactionInfo.java
+++ b/src/java/org/apache/cassandra/io/CompactionInfo.java
@@ -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;
}
diff --git a/src/java/org/apache/cassandra/io/CompactionIterator.java b/src/java/org/apache/cassandra/io/CompactionIterator.java
index 892295fd57..744e009082 100644
--- a/src/java/org/apache/cassandra/io/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/io/CompactionIterator.java
@@ -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 rows = new ArrayList();
- 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 sstables, CompactionController controller) throws IOException
+ public CompactionIterator(CompactionType type, Iterable 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;
diff --git a/src/java/org/apache/cassandra/io/CompactionType.java b/src/java/org/apache/cassandra/io/CompactionType.java
new file mode 100644
index 0000000000..5aefc86c02
--- /dev/null
+++ b/src/java/org/apache/cassandra/io/CompactionType.java
@@ -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;
+ }
+}
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
index 4c2f13c195..78f773d076 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
@@ -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());
}
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 720a947073..41dfba437a 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -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 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 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 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 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();
}
/**
diff --git a/src/resources/org/apache/cassandra/cli/CliHelp.yaml b/src/resources/org/apache/cassandra/cli/CliHelp.yaml
new file mode 100644
index 0000000000..2f8e38e7af
--- /dev/null
+++ b/src/resources/org/apache/cassandra/cli/CliHelp.yaml
@@ -0,0 +1,1150 @@
+# Help file for online commands in Yaml.
+
+banner: |
+ Welcome to the Cassandra CLI.
+
+ Type 'help;' or '?' for help.
+ Type 'quit;' or 'exit;' to quit.
+
+help: |
+ Getting around:
+ ? Display this help.
+ help; Display this help.
+ help