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@1095492 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-04-20 19:50:50 +00:00
parent 20cdc0f475
commit 2d95cb4957
32 changed files with 230 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGES.txt
View File

@ -57,6 +57,10 @@
* use 64KB flush buffer instead of in_memory_compaction_limit (CASSANDRA-2463)
* fix duplicate results from CFS.scan (CASSANDRA-2406)
* avoid caching token-only decoratedkeys (CASSANDRA-2416)
* count a row deletion as one operation towards memtable threshold
(CASSANDRA-2519)
* fixes for verifying destination availability under hinted conditions
so UE can be thrown intead of timing out (CASSANDRA-2514)
0.7.4

4
conf/cassandra-env.sh
View File

@ -92,8 +92,8 @@ JMX_PORT="7199"
JVM_OPTS="$JVM_OPTS -ea"
# add the jamm javaagent
java_version=`java -version 2>&1`
if [[ $java_version != *OpenJDK* ]]
check_openjdk=$(java -version 2>&1 | awk '{if (NR == 2) {print $1}}')
if [ "$check_openjdk" != "OpenJDK" ]
then
JVM_OPTS="$JVM_OPTS -javaagent:$CASSANDRA_HOME/lib/jamm-0.2.1.jar"
fi

6
contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
View File

@ -179,8 +179,8 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface
AbstractType default_validator = null;
try
{
comparator = FBUtilities.getInstance(cfDef.comparator_type, "comparator");
default_validator = FBUtilities.getInstance(cfDef.default_validation_class, "validator");
comparator = FBUtilities.getComparator(cfDef.comparator_type);
default_validator = FBUtilities.getComparator(cfDef.default_validation_class);
}
catch (ConfigurationException e)
{
@ -202,7 +202,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface
AbstractType validator = null;
try
{
validator = FBUtilities.getInstance(cd.getValidation_class(), "validator");
validator = FBUtilities.getComparator(cd.getValidation_class());
validators.put(cd.name, validator);
}
catch (ConfigurationException e)

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

@ -17,6 +17,7 @@
import exceptions
import datetime
import time
import connection
import marshal

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

@ -19,6 +19,7 @@ from cursor import Cursor
from cassandra import Cassandra
from thrift.transport import TTransport, TSocket
from thrift.protocol import TBinaryProtocol
from cql.cassandra.ttypes import AuthenticationRequest
class Connection(object):
@ -62,7 +63,7 @@ class Connection(object):
def close(self):
if not self.open_socket:
raise InternalError("Connection has been closed.")
return
self.transport.close()
self.open_socket = False
@ -75,9 +76,11 @@ class Connection(object):
return
def rollback(self):
from cql import NotSupportedError
raise NotSupportedError("Rollback functionality not present in Cassandra.")
def cursor(self):
from cql import ProgrammingError
if not self.open_socket:
raise InternalError("Connection has been closed.")
raise ProgrammingError("Connection has been closed.")
return Cursor(self)

11
drivers/py/cql/cursor.py
View File

@ -62,6 +62,7 @@ class Cursor:
def prepare(self, query, params):
prepared_query = prepare(query, params)
self._schema_update_needed = False
# Snag the keyspace or column family and stash it for later use in
# decoding columns. These regexes don't match every query, but the
@ -78,9 +79,7 @@ class Cursor:
# 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()
self._schema_update_needed = True
return prepared_query
def __get_schema(self):
@ -110,6 +109,9 @@ class Cursor:
def execute(self, cql_query, params={}):
self.__checksock()
self.rs_idx = 0
self.rowcount = 0
self.description = None
try:
prepared_q = self.prepare(cql_query, params)
except KeyError, e:
@ -132,6 +134,9 @@ class Cursor:
except TApplicationException, tapp:
raise cql.InternalError("Internal application error")
if self._schema_update_needed and isinstance(self.decoder, SchemaDecoder):
self.decoder.schema = self.__get_schema()
if response.type == CqlResultType.ROWS:
self.result = ResultSet(response.rows,
self._query_ks,

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

@ -26,9 +26,8 @@ class SchemaDecoder(object):
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]
if keyspace in self.schema and column_family in self.schema[keyspace]:
return self.schema[keyspace][column_family]
return None
def __comparator_for(self, keyspace, column_family):

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

@ -66,11 +66,12 @@ def unmarshal_utf8(bytestr):
def unmarshal_int(bytestr):
return decode_bigint(bytestr)
def unmarshal_long(bytestr):
if _have_struct:
if _have_struct:
def unmarshal_long(bytestr):
return _long_packer.unpack(bytestr)[0]
else:
return unpack(">q", bytestr)[0]
else:
def unmarshal_long(bytestr):
return struct.unpack(">q", bytestr)[0]
def unmarshal_uuid(bytestr):
return UUID(bytes=bytestr)

94
drivers/py/cqlsh
View File

@ -48,12 +48,13 @@ def startswith(words, text):
class Shell(cmd.Cmd):
default_prompt = "cqlsh> "
continue_prompt = " ... "
def __init__(self, hostname, port, color=False, username=None,
password=None):
cmd.Cmd.__init__(self)
self.conn = cql.connect(hostname, port, user=username, password=password)
self.cursor = self.conn.cursor()
if os.path.exists(HISTORY):
readline.read_history_file(HISTORY)
@ -64,12 +65,33 @@ class Shell(cmd.Cmd):
self.statement = StringIO()
self.color = color
self.in_comment = False
def reset_statement(self):
self.set_prompt(Shell.default_prompt)
self.statement.truncate(0)
def get_statement(self, line):
if self.in_comment:
if "*/" in line:
fragment = line[line.index("*/")+2:]
if fragment.strip():
line = fragment
self.in_comment = False
else:
self.in_comment = False
self.set_prompt(Shell.default_prompt)
return None
else:
return None
if "/*" in line and (not self.in_comment):
self.in_comment = True
self.set_prompt(Shell.continue_prompt)
if line.lstrip().index("/*") != 0:
self.statement.write(line[:line.lstrip().index("/*")])
return None
self.statement.write("%s\n" % line)
if not line.endswith(";"):
@ -82,25 +104,31 @@ class Shell(cmd.Cmd):
self.reset_statement()
def default(self, arg):
if not arg.strip(): return
statement = self.get_statement(arg)
def scrub_oneline_comments(s):
res = re.sub(r'\/\*.*\*\/', '', s)
res = re.sub(r'--.*$', '', res)
return res
input = scrub_oneline_comments(arg)
if not input.strip(): return
statement = self.get_statement(input)
if not statement: return
cursor = self.conn.cursor()
cursor.execute(statement)
if isinstance(cursor.result, ResultSet):
for row in cursor.result.rows:
self.printout(row.key, BLUE, False)
for column in row.columns:
self.cursor.execute(statement)
if isinstance(self.cursor.result, ResultSet):
for x in range(self.cursor.rowcount):
row = self.cursor.fetchone()
self.printout(repr(row[0]), BLUE, False)
for (i, value) in enumerate(row[1:]):
name = self.cursor.description[i+1][0]
self.printout(" | ", newline=False)
# XXX: repr() is better than trying to print binary
self.printout(repr(column.name), MAGENTA, False)
self.printout(repr(name), MAGENTA, False)
self.printout(",", newline=False)
self.printout(repr(column.value), YELLOW, False)
self.printout(repr(value), YELLOW, False)
self.printout("")
else:
if cursor.result: print cursor.result[0]
if self.cursor.result: print self.cursor.result[0]
def emptyline(self):
pass
@ -109,19 +137,19 @@ class Shell(cmd.Cmd):
keywords = ('FIRST', 'REVERSED', 'FROM', 'WHERE', 'KEY')
return startswith(keywords, text.upper())
complete_SELECT = complete_select
def complete_update(self, text, line, begidx, endidx):
keywords = ('WHERE', 'KEY', 'SET')
return startswith(keywords, text.upper())
complete_UPDATE = complete_update
def complete_create(self, text, line, begidx, endidx):
words = line.split()
if len(words) < 3:
return startswith(['COLUMNFAMILY', 'KEYSPACE'], text.upper())
common = ['WITH', 'AND']
if words[1].upper() == 'COLUMNFAMILY':
types = startswith(CQLTYPES, text)
keywords = startswith(('KEY', 'PRIMARY'), text.upper())
@ -141,12 +169,12 @@ class Shell(cmd.Cmd):
"memtable_operations_in_millions",
"replicate_on_write"), text)
return startswith(common, text.upper()) + types + keywords + props
if words[1].upper() == 'KEYSPACE':
props = ("replication_factor", "strategy_options", "strategy_class")
return startswith(common, text.upper()) + startswith(props, text)
complete_CREATE = complete_create
def complete_drop(self, text, line, begidx, endidx):
words = line.split()
if len(words) < 3:
@ -161,27 +189,27 @@ class Shell(cmd.Cmd):
def set_prompt(self, prompt):
if sys.stdin.isatty():
self.prompt = prompt
def do_EOF(self, arg):
if sys.stdin.isatty(): print
self.do_exit(None)
def do_exit(self, arg):
sys.exit()
do_quit = do_exit
def printout(self, text, color=None, newline=True, out=sys.stdout):
if not color or not self.color:
out.write(text)
else:
out.write(color % text)
if newline:
out.write("\n");
def printerr(self, text, color=None, newline=True):
self.printout(text, color, newline, sys.stderr)
if __name__ == '__main__':
parser = OptionParser(usage = "Usage: %prog [host [port]]")
parser.add_option("-C",
@ -191,9 +219,9 @@ if __name__ == '__main__':
parser.add_option("-u", "--username", help="Authenticate as user.")
parser.add_option("-p", "--password", help="Authenticate using password.")
(options, arguments) = parser.parse_args()
hostname = len(arguments) > 0 and arguments[0] or "localhost"
if len(arguments) > 1:
try:
port = int(arguments[1])
@ -203,8 +231,8 @@ if __name__ == '__main__':
sys.exit(1)
else:
port = 9160
shell = Shell(hostname,
port,
color=options.color,
@ -223,4 +251,4 @@ if __name__ == '__main__':
print
except Exception, err:
shell.printerr("Exception: %s" % err, color=RED)

32
examples/hadoop_word_count/src/WordCountSetup.java
View File

@ -51,13 +51,19 @@ public class WordCountSetup
// text0: no rows
// text1: 1 row, 1 word
c = new Column(ByteBufferUtil.bytes("text1"), ByteBufferUtil.bytes("word1"), System.currentTimeMillis());
c = new Column()
.setName(ByteBufferUtil.bytes("text1"))
.setValue(ByteBufferUtil.bytes("word1"))
.setTimestamp(System.currentTimeMillis());
mutationMap = getMutationMap(ByteBufferUtil.bytes("key0"), WordCount.COLUMN_FAMILY, c);
client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
logger.info("added text1");
// text1: 1 row, 2 word
c = new Column(ByteBufferUtil.bytes("text2"), ByteBufferUtil.bytes("word1 word2"), System.currentTimeMillis());
c = new Column()
.setName(ByteBufferUtil.bytes("text2"))
.setValue(ByteBufferUtil.bytes("word1 word2"))
.setTimestamp(System.currentTimeMillis());
mutationMap = getMutationMap(ByteBufferUtil.bytes("key0"), WordCount.COLUMN_FAMILY, c);
client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
logger.info("added text2");
@ -66,7 +72,10 @@ public class WordCountSetup
mutationMap = new HashMap<ByteBuffer,Map<String,List<Mutation>>>();
for (int i=0; i<1000; i++)
{
c = new Column(ByteBufferUtil.bytes("text3"), ByteBufferUtil.bytes("word1"), System.currentTimeMillis());
c = new Column()
.setName(ByteBufferUtil.bytes("text3"))
.setValue(ByteBufferUtil.bytes("word1"))
.setTimestamp(System.currentTimeMillis());
addToMutationMap(mutationMap, ByteBufferUtil.bytes("key" + i), WordCount.COLUMN_FAMILY, c);
}
client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
@ -96,19 +105,20 @@ public class WordCountSetup
mutationMap.put(key, cfMutation);
}
private static void setupKeyspace(Cassandra.Iface client) throws TException, InvalidRequestException
{
private static void setupKeyspace(Cassandra.Iface client) throws TException, InvalidRequestException, SchemaDisagreementException {
List<CfDef> cfDefList = new ArrayList<CfDef>();
CfDef input = new CfDef(WordCount.KEYSPACE, WordCount.COLUMN_FAMILY);
input.setComparator_type("AsciiType");
input.setDefault_validation_class("AsciiType");
cfDefList.add(input);
input.setComparator_type("AsciiType");
input.setDefault_validation_class("AsciiType");
cfDefList.add(input);
CfDef output = new CfDef(WordCount.KEYSPACE, WordCount.OUTPUT_COLUMN_FAMILY);
output.setComparator_type("AsciiType");
output.setDefault_validation_class("AsciiType");
output.setComparator_type("AsciiType");
output.setDefault_validation_class("AsciiType");
cfDefList.add(output);
client.system_add_keyspace(new KsDef(WordCount.KEYSPACE, "org.apache.cassandra.locator.SimpleStrategy", 1, cfDefList));
KsDef ksDef = new KsDef(WordCount.KEYSPACE, "org.apache.cassandra.locator.SimpleStrategy", cfDefList);
ksDef.putToStrategy_options("replication_factor", "1");
client.system_add_keyspace(ksDef);
int magnitude = client.describe_ring(WordCount.KEYSPACE).size();
try
{

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

@ -487,7 +487,7 @@ public class CliClient
sessionState.out.println("Returned " + columns.size() + " results.");
}
private AbstractType getFormatTypeForColumn(String compareWith)
private AbstractType getFormatType(String compareWith)
{
Function function;
@ -597,7 +597,7 @@ public class CliClient
// .getText() will give us <type>
String typeName = CliUtils.unescapeSQLString(typeTree.getText());
// building AbstractType from <type>
AbstractType valueValidator = getFormatTypeForColumn(typeName);
AbstractType valueValidator = getFormatType(typeName);
// setting value for output
valueAsString = valueValidator.getString(ByteBuffer.wrap(columnValue));
@ -1597,8 +1597,9 @@ public class CliClient
String leftSpace = " ";
String columnLeftSpace = leftSpace + " ";
AbstractType columnNameValidator = getFormatTypeForColumn(isSuper ? cf_def.subcomparator_type
: cf_def.comparator_type);
String compareWith = isSuper ? cf_def.subcomparator_type
: cf_def.comparator_type;
AbstractType columnNameValidator = getFormatType(compareWith);
sessionState.out.println(leftSpace + "Column Metadata:");
for (ColumnDef columnDef : cf_def.getColumn_metadata())
@ -1927,7 +1928,7 @@ public class CliClient
private ByteBuffer columnNameAsBytes(String column, CfDef columnFamilyDef)
{
String comparatorClass = columnFamilyDef.comparator_type;
return getBytesAccordingToType(column, getFormatTypeForColumn(comparatorClass));
return getBytesAccordingToType(column, getFormatType(comparatorClass));
}
/**
@ -1980,7 +1981,7 @@ public class CliClient
comparatorClass = "BytesType";
}
return getBytesAccordingToType(superColumn, getFormatTypeForColumn(comparatorClass));
return getBytesAccordingToType(superColumn, getFormatType(comparatorClass));
}
/**
@ -2025,7 +2026,7 @@ public class CliClient
try
{
String validationClass = columnDefinition.getValidation_class();
return getBytesAccordingToType(columnValue, getFormatTypeForColumn(validationClass));
return getBytesAccordingToType(columnValue, getFormatType(validationClass));
}
catch (Exception e)
{
@ -2054,13 +2055,13 @@ public class CliClient
if (Arrays.equals(nameInBytes, columnNameInBytes))
{
return getFormatTypeForColumn(columnDefinition.getValidation_class());
return getFormatType(columnDefinition.getValidation_class());
}
}
if (defaultValidator != null && !defaultValidator.isEmpty())
{
return getFormatTypeForColumn(defaultValidator);
return getFormatType(defaultValidator);
}
return null;
@ -2254,7 +2255,7 @@ public class CliClient
{
AbstractType validator;
String columnFamilyName = columnFamilyDef.getName();
AbstractType keyComparator = this.cfKeysComparators.get(columnFamilyName);
AbstractType keyComparator = getKeyComparatorForCF(columnFamilyName);
for (KeySlice ks : slices)
{
@ -2321,14 +2322,14 @@ public class CliClient
private String formatSubcolumnName(String keyspace, String columnFamily, ByteBuffer name)
throws NotFoundException, TException, IllegalAccessException, InstantiationException, NoSuchFieldException
{
return getFormatTypeForColumn(getCfDef(keyspace,columnFamily).subcomparator_type).getString(name);
return getFormatType(getCfDef(keyspace, columnFamily).subcomparator_type).getString(name);
}
// retuns column name in human-readable format
private String formatColumnName(String keyspace, String columnFamily, ByteBuffer name)
throws NotFoundException, TException, IllegalAccessException, InstantiationException, NoSuchFieldException
{
return getFormatTypeForColumn(getCfDef(keyspace, columnFamily).comparator_type).getString(name);
return getFormatType(getCfDef(keyspace, columnFamily).comparator_type).getString(name);
}
private ByteBuffer getColumnName(String columnFamily, Tree columnTree)
@ -2352,8 +2353,21 @@ public class CliClient
String key = CliUtils.unescapeSQLString(keyTree.getText());
AbstractType keyComparator = this.cfKeysComparators.get(columnFamily);
return getBytesAccordingToType(key, keyComparator);
return getBytesAccordingToType(key, getKeyComparatorForCF(columnFamily));
}
private AbstractType getKeyComparatorForCF(String columnFamily)
{
AbstractType keyComparator = cfKeysComparators.get(columnFamily);
if (keyComparator == null)
{
String defaultValidationClass = getCfDef(columnFamily).getKey_validation_class();
assert defaultValidationClass != null;
keyComparator = getFormatType(defaultValidationClass);
}
return keyComparator;
}
private static class KsDefNamesComparator implements Comparator<KsDef>
@ -2417,7 +2431,7 @@ public class CliClient
String defaultValidator = cfdef.default_validation_class;
if (defaultValidator != null && !defaultValidator.isEmpty())
{
return (getFormatTypeForColumn(defaultValidator) instanceof CounterColumnType);
return (getFormatType(defaultValidator) instanceof CounterColumnType);
}
return false;
}

4
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -208,7 +208,9 @@ public class Memtable implements Comparable<Memtable>, IFlushable
private void resolve(DecoratedKey key, ColumnFamily cf)
{
currentThroughput.addAndGet(cf.size());
currentOperations.addAndGet(cf.getColumnCount());
currentOperations.addAndGet((cf.getColumnCount() == 0)
? cf.isMarkedForDelete() ? 1 : 0
: cf.getColumnCount());
ColumnFamily oldCf = columnFamilies.putIfAbsent(key, cf);
if (oldCf == null)

6
src/java/org/apache/cassandra/db/marshal/AbstractCommutativeType.java
View File

@ -24,6 +24,7 @@ import java.nio.ByteBuffer;
import org.apache.cassandra.db.Column;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.utils.ByteBufferUtil;
public abstract class AbstractCommutativeType extends AbstractType<Long>
{
@ -37,6 +38,11 @@ public abstract class AbstractCommutativeType extends AbstractType<Long>
return CounterContext.instance().total(bytes);
}
public ByteBuffer decompose(Long value)
{
return ByteBufferUtil.bytes(value);
}
/**
* create commutative column
*/

2
src/java/org/apache/cassandra/db/marshal/AbstractType.java
View File

@ -93,6 +93,8 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
}
public abstract T compose(ByteBuffer bytes);
public abstract ByteBuffer decompose(T value);
/** get a string representation of a particular type. */
public abstract String toString(T t);

7
src/java/org/apache/cassandra/db/marshal/AsciiType.java
View File

@ -61,9 +61,14 @@ public class AsciiType extends AbstractType<String>
return getString(bytes);
}
public ByteBuffer decompose(String value)
{
return ByteBufferUtil.bytes(value, Charsets.US_ASCII);
}
public ByteBuffer fromString(String source)
{
return ByteBufferUtil.bytes(source, Charsets.US_ASCII);
return decompose(source);
}
public void validate(ByteBuffer bytes) throws MarshalException

7
src/java/org/apache/cassandra/db/marshal/BytesType.java
View File

@ -36,12 +36,17 @@ public class BytesType extends AbstractType<ByteBuffer>
{
return bytes.duplicate();
}
public ByteBuffer decompose(ByteBuffer value)
{
return value;
}
public int compare(ByteBuffer o1, ByteBuffer o2)
{
return BytesType.bytesCompare(o1, o2);
}
public static int bytesCompare(ByteBuffer o1, ByteBuffer o2)
{
if(null == o1){

7
src/java/org/apache/cassandra/db/marshal/IntegerType.java
View File

@ -62,6 +62,11 @@ public final class IntegerType extends AbstractType<BigInteger>
return new BigInteger(ByteBufferUtil.getArray(bytes));
}
public ByteBuffer decompose(BigInteger value)
{
return ByteBuffer.wrap(value.toByteArray());
}
public int compare(ByteBuffer lhs, ByteBuffer rhs)
{
int lhsLen = lhs.remaining();
@ -152,7 +157,7 @@ public final class IntegerType extends AbstractType<BigInteger>
throw new MarshalException(String.format("unable to make int from '%s'", source), e);
}
return ByteBuffer.wrap(integerType.toByteArray());
return decompose(integerType);
}
public void validate(ByteBuffer bytes) throws MarshalException

7
src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java
View File

@ -38,6 +38,11 @@ public class LexicalUUIDType extends AbstractType<UUID>
return UUIDGen.getUUID(bytes);
}
public ByteBuffer decompose(UUID value)
{
return ByteBuffer.wrap(UUIDGen.decompose(value));
}
public int compare(ByteBuffer o1, ByteBuffer o2)
{
if (o1.remaining() == 0)
@ -78,7 +83,7 @@ public class LexicalUUIDType extends AbstractType<UUID>
try
{
return ByteBuffer.wrap(UUIDGen.decompose(UUID.fromString(source)));
return decompose(UUID.fromString(source));
}
catch (IllegalArgumentException e)
{

5
src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java
View File

@ -43,6 +43,11 @@ public class LocalByPartionerType<T extends Token> extends AbstractType<ByteBuff
throw new UnsupportedOperationException("You can't do this with a local partitioner.");
}
public ByteBuffer decompose(ByteBuffer bytes)
{
throw new UnsupportedOperationException("You can't do this with a local partitioner.");
}
public String getString(ByteBuffer bytes)
{
return ByteBufferUtil.bytesToHex(bytes);

7
src/java/org/apache/cassandra/db/marshal/LongType.java
View File

@ -36,6 +36,11 @@ public class LongType extends AbstractType<Long>
return ByteBufferUtil.toLong(bytes);
}
public ByteBuffer decompose(Long value)
{
return ByteBufferUtil.bytes(value);
}
public int compare(ByteBuffer o1, ByteBuffer o2)
{
if (o1.remaining() == 0)
@ -91,7 +96,7 @@ public class LongType extends AbstractType<Long>
throw new MarshalException(String.format("unable to make long from '%s'", source), e);
}
return ByteBufferUtil.bytes(longType);
return decompose(longType);
}
public void validate(ByteBuffer bytes) throws MarshalException

9
src/java/org/apache/cassandra/db/marshal/TimeUUIDType.java
View File

@ -33,8 +33,8 @@ import org.apache.commons.lang.time.DateUtils;
public class TimeUUIDType extends AbstractType<UUID>
{
public static final TimeUUIDType instance = new TimeUUIDType();
static final Pattern regexPattern = Pattern.compile("[A-Fa-f0-9]{8}\\-[A-Fa-f0-9]{4}\\-[A-Fa-f0-9]{4}\\-[A-Fa-f0-9]{4}\\-[A-Fa-f0-9]{12}");
static final String[] iso8601Patterns = new String[] {
"yyyy-MM-dd HH:mm",
@ -56,6 +56,11 @@ public class TimeUUIDType extends AbstractType<UUID>
return UUIDGen.getUUID(bytes);
}
public ByteBuffer decompose(UUID value)
{
return ByteBuffer.wrap(UUIDGen.decompose(value));
}
public int compare(ByteBuffer o1, ByteBuffer o2)
{
if (o1.remaining() == 0)
@ -139,7 +144,7 @@ public class TimeUUIDType extends AbstractType<UUID>
try
{
uuid = UUID.fromString(source);
idBytes = ByteBuffer.wrap(UUIDGen.decompose(uuid));
idBytes = decompose(uuid);
}
catch (IllegalArgumentException e)
{

10
src/java/org/apache/cassandra/db/marshal/UTF8Type.java
View File

@ -23,6 +23,7 @@ package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import com.google.common.base.Charsets;
import org.apache.cassandra.utils.ByteBufferUtil;
public class UTF8Type extends AbstractType<String>
@ -36,6 +37,11 @@ public class UTF8Type extends AbstractType<String>
return getString(bytes);
}
public ByteBuffer decompose(String value)
{
return ByteBufferUtil.bytes(value, Charsets.UTF_8);
}
public int compare(ByteBuffer o1, ByteBuffer o2)
{
return BytesType.bytesCompare(o1, o2);
@ -60,9 +66,9 @@ public class UTF8Type extends AbstractType<String>
public ByteBuffer fromString(String source)
{
return ByteBufferUtil.bytes(source);
return decompose(source);
}
public void validate(ByteBuffer bytes) throws MarshalException
{
if (!UTF8Validator.validate(bytes.slice()))

5
src/java/org/apache/cassandra/db/marshal/UUIDType.java
View File

@ -204,6 +204,11 @@ public class UUIDType extends AbstractType<UUID>
return uuid.toString();
}
public ByteBuffer decompose(UUID value)
{
return ByteBuffer.wrap(UUIDGen.decompose(value));
}
@Override
public ByteBuffer fromString(String source) throws MarshalException
{

10
src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java
View File

@ -102,10 +102,12 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan
dcEndpoints.put(dc, new AtomicInteger());
for (InetAddress destination : hintedEndpoints.keySet())
{
assert writeEndpoints.contains(destination);
// figure out the destination dc
String destinationDC = snitch.getDatacenter(destination);
dcEndpoints.get(destinationDC).incrementAndGet();
if (writeEndpoints.contains(destination))
{
// figure out the destination dc
String destinationDC = snitch.getDatacenter(destination);
dcEndpoints.get(destinationDC).incrementAndGet();
}
}
// Throw exception if any of the DC doesn't have livenodes to accept write.

4
src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java
View File

@ -84,9 +84,9 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler
public void assureSufficientLiveNodes() throws UnavailableException
{
int liveNodes = 0;
for (InetAddress destination : writeEndpoints)
for (InetAddress destination : hintedEndpoints.keySet())
{
if (localdc.equals(snitch.getDatacenter(destination)))
if (localdc.equals(snitch.getDatacenter(destination)) && writeEndpoints.contains(destination))
liveNodes++;
}

1
src/java/org/apache/cassandra/service/WriteResponseHandler.java
View File

@ -100,6 +100,7 @@ public class WriteResponseHandler extends AbstractWriteResponseHandler
// ensure there are blockFor distinct living nodes (hints are ok).
if (hintedEndpoints.keySet().size() < responses.get())
throw new UnavailableException();
return;
}
// count destinations that are part of the desired target set

26
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -564,7 +564,6 @@ public class FBUtilities
/**
* Constructs an instance of the given class, which must have a no-arg constructor.
* TODO: Similar method for our 'instance member' singleton pattern would be nice.
* @param classname Fully qualified classname.
* @param readable Descriptive noun for the role the class plays.
* @throws ConfigurationException If the class cannot be found.
@ -596,31 +595,6 @@ public class FBUtilities
}
}
public static <T> T getInstance(String classname, String readable) throws ConfigurationException
{
Class cls = classForName(classname, readable);
T rval = null;
try
{
rval = (T) cls.getDeclaredMethod("getInstance").invoke(new Object[] {null, null});
}
catch (NoSuchMethodException e)
{
throw new ConfigurationException("Class does not have the getInstance method with no arguments");
}
catch (InvocationTargetException e)
{
throw new ConfigurationException(String.format("Could not call method getInstance on %s class %s", readable, classname));
}
catch (IllegalAccessException e)
{
throw new ConfigurationException(String.format("Could not call method getInstance on %s class %s", readable, classname));
}
return rval;
}
public static <T extends Comparable> SortedSet<T> singleton(T column)
{
return new TreeSet<T>(Arrays.asList(column));

16
test/distributed/org/apache/cassandra/MovementTest.java
View File

@ -50,16 +50,12 @@ public class MovementTest extends TestBase
private static Map<ByteBuffer,List<ColumnOrSuperColumn>> insertBatch(Cassandra.Client client) throws Exception
{
final int N = 1000;
Column col1 = new Column(
ByteBufferUtil.bytes("c1"),
ByteBufferUtil.bytes("v1"),
0
);
Column col2 = new Column(
ByteBufferUtil.bytes("c2"),
ByteBufferUtil.bytes("v2"),
0
);
Column col1 = new Column(ByteBufferUtil.bytes("c1"))
.setValue(ByteBufferUtil.bytes("v1"))
.setTimestamp(0);
Column col2 = new Column(ByteBufferUtil.bytes("c2"))
.setValue(ByteBufferUtil.bytes("v2"))
.setTimestamp(0);
// build N rows
Map<ByteBuffer,List<ColumnOrSuperColumn>> rows = new HashMap<ByteBuffer, List<ColumnOrSuperColumn>>();

184
test/distributed/org/apache/cassandra/MutationTest.java
View File

@ -180,190 +180,6 @@ public class MutationTest extends TestBase
}
}
protected void insert(Cassandra.Client client, ByteBuffer key, String cf, String name, String value, long timestamp, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
Column col = new Column(
ByteBufferUtil.bytes(name),
ByteBufferUtil.bytes(value),
timestamp
);
client.insert(key, new ColumnParent(cf), col, cl);
}
protected Column getColumn(Cassandra.Client client, ByteBuffer key, String cf, String col, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException
{
ColumnPath cpath = new ColumnPath(cf);
cpath.setColumn(col.getBytes());
return client.get(key, cpath, cl).column;
}
protected class Get extends RetryingAction
{
public Get(Cassandra.Client client, String cf, ByteBuffer key)
{
super(client, cf, key);
}
public void tryPerformAction(ConsistencyLevel cl) throws Exception
{
assertColumnEqual(name, value, timestamp, getColumn(client, key, cf, name, cl));
}
}
protected class Insert extends RetryingAction
{
public Insert(Cassandra.Client client, String cf, ByteBuffer key)
{
super(client, cf, key);
}
public void tryPerformAction(ConsistencyLevel cl) throws Exception
{
insert(client, key, cf, name, value, timestamp, cl);
}
}
/** Performs an action repeatedly until timeout, success or failure. */
protected abstract class RetryingAction
{
protected Cassandra.Client client;
protected String cf;
protected ByteBuffer key;
protected String name;
protected String value;
protected long timestamp;
private Set<Class<Exception>> expected = new HashSet<Class<Exception>>();
private long timeout = StorageService.RING_DELAY;
public RetryingAction(Cassandra.Client client, String cf, ByteBuffer key)
{
this.client = client;
this.cf = cf;
this.key = key;
this.timestamp = 0;
}
public RetryingAction name(String name)
{
this.name = name; return this;
}
/** The value to expect for the return column, or null to expect the column to be missing. */
public RetryingAction value(String value)
{
this.value = value; return this;
}
/** The total time to allow before failing. */
public RetryingAction timeout(long timeout)
{
this.timeout = timeout; return this;
}
/** The expected timestamp of the returned column. */
public RetryingAction timestamp(long timestamp)
{
this.timestamp = timestamp; return this;
}
/** The exception classes that indicate success. */
public RetryingAction expecting(Class... tempExceptions)
{
this.expected.clear();
for (Class exclass : tempExceptions)
expected.add((Class<Exception>)exclass);
return this;
}
public void perform(ConsistencyLevel cl) throws AssertionError
{
long deadline = System.currentTimeMillis() + timeout;
int attempts = 0;
String template = "%s for " + this + " after %d attempt(s) with %d ms to spare.";
Exception e = null;
while(deadline > System.currentTimeMillis())
{
try
{
attempts++;
tryPerformAction(cl);
logger.info(String.format(template, "Succeeded", attempts, deadline - System.currentTimeMillis()));
return;
}
catch (Exception ex)
{
e = ex;
if (!expected.contains(ex.getClass()))
continue;
logger.info(String.format(template, "Caught expected exception: " + e, attempts, deadline - System.currentTimeMillis()));
return;
}
}
String err = String.format(template, "Caught unexpected: " + e, attempts, deadline - System.currentTimeMillis());
logger.error(err);
throw new AssertionError(err);
}
public String toString()
{
return this.getClass() + "(" + key + "," + name + ")";
}
protected abstract void tryPerformAction(ConsistencyLevel cl) throws Exception;
}
protected List<ColumnOrSuperColumn> get_slice(Cassandra.Client client, ByteBuffer key, String cf, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
SlicePredicate sp = new SlicePredicate();
sp.setSlice_range(
new SliceRange(
ByteBuffer.wrap(new byte[0]),
ByteBuffer.wrap(new byte[0]),
false,
1000
)
);
return client.get_slice(key, new ColumnParent(cf), sp, cl);
}
protected void assertColumnEqual(String name, String value, long timestamp, Column col)
{
assertEquals(ByteBufferUtil.bytes(name), col.name);
assertEquals(ByteBufferUtil.bytes(value), col.value);
assertEquals(timestamp, col.timestamp);
}
protected List<InetAddress> endpointsForKey(InetAddress seed, ByteBuffer key, String keyspace)
throws IOException
{
RingCache ring = new RingCache(keyspace, new RandomPartitioner(), seed.getHostAddress(), 9160);
List<InetAddress> privateendpoints = ring.getEndpoint(key);
List<InetAddress> endpoints = new ArrayList<InetAddress>();
for (InetAddress endpoint : privateendpoints)
{
endpoints.add(controller.getPublicHost(endpoint));
}
return endpoints;
}
protected InetAddress nonEndpointForKey(InetAddress seed, ByteBuffer key, String keyspace)
throws IOException
{
List<InetAddress> endpoints = endpointsForKey(seed, key, keyspace);
for (InetAddress host : controller.getHosts())
{
if (!endpoints.contains(host))
{
return host;
}
}
return null;
}
protected ByteBuffer newKey()
{
return ByteBufferUtil.bytes(String.format("test.key.%d", System.currentTimeMillis()));

8
test/distributed/org/apache/cassandra/TestBase.java
View File

@ -149,11 +149,9 @@ public abstract class TestBase
protected void insert(Cassandra.Client client, ByteBuffer key, String cf, String name, String value, long timestamp, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
Column col = new Column(
ByteBuffer.wrap(name.getBytes()),
ByteBuffer.wrap(value.getBytes()),
timestamp
);
Column col = new Column(ByteBuffer.wrap(name.getBytes()))
.setValue(ByteBuffer.wrap(value.getBytes()))
.setTimestamp(timestamp);
client.insert(key, new ColumnParent(cf), col, cl);
}

5
tools/py_stress/stress.py
View File

@ -185,7 +185,10 @@ def make_keyspaces():
colms = [ColumnDef(name='C1', validation_class='UTF8Type', index_type=IndexType.KEYS_BITMAP)]
cfams = [CfDef(keyspace='Keyspace1', name='Standard1', column_metadata=colms),
CfDef(keyspace='Keyspace1', name='Super1', column_type='Super')]
keyspace = KsDef(name='Keyspace1', strategy_class='org.apache.cassandra.locator.SimpleStrategy', replication_factor=options.replication, cf_defs=cfams)
keyspace = KsDef(name='Keyspace1',
strategy_class='org.apache.cassandra.locator.SimpleStrategy',
strategy_options={'replication_factor': str(options.replication)},
cf_defs=cfams)
client = get_client(nodes[0], options.port)
client.transport.open()
try:

2
tools/stress/src/org/apache/cassandra/stress/operations/Inserter.java
View File

@ -49,7 +49,7 @@ public class Inserter extends Operation
{
String columnName = ("C" + Integer.toString(i));
ByteBuffer columnValue = values.get(i % values.size());
columns.add(new Column(ByteBufferUtil.bytes(columnName), columnValue, System.currentTimeMillis()));
columns.add(new Column(ByteBufferUtil.bytes(columnName)).setValue(columnValue).setTimestamp(System.currentTimeMillis()));
}
if (session.getColumnFamilyType() == ColumnFamilyType.Super)