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

convert thrift to byte keys. Patch by Stu Hood, reviewed by Gary Dusbabek. CASSANDRA-767 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@934155 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Dusbabek 2010-04-14 19:53:23 +00:00
parent 434564d1f9
commit f3a4e1736b
14 changed files with 378 additions and 321 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -60,7 +60,7 @@ public class CassandraStorage extends LoadFunc
// load the next pair
if (!reader.nextKeyValue())
return null;
String key = (String)reader.getCurrentKey();
byte[] key = (byte[])reader.getCurrentKey();
SortedMap<byte[],IColumn> cf = (SortedMap<byte[],IColumn>)reader.getCurrentValue();
assert key != null && cf != null;

4
contrib/word_count/src/WordCount.java
View File

@ -63,13 +63,13 @@ public class WordCount extends Configured implements Tool
System.exit(0);
}
public static class TokenizerMapper extends Mapper<String, SortedMap<byte[], IColumn>, Text, IntWritable>
public static class TokenizerMapper extends Mapper<byte[], SortedMap<byte[], IColumn>, Text, IntWritable>
{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
private String columnName;
public void map(String key, SortedMap<byte[], IColumn> columns, Context context) throws IOException, InterruptedException
public void map(byte[] key, SortedMap<byte[], IColumn> columns, Context context) throws IOException, InterruptedException
{
IColumn column = columns.get(columnName.getBytes());
if (column == null)

38
interface/cassandra.thrift
View File

@ -46,7 +46,7 @@ namespace rb CassandraThrift
# for every edit that doesn't result in a change to major/minor.
#
# See the Semantic Versioning Specification (SemVer) http://semver.org.
const string VERSION = "3.0.0"
const string VERSION = "4.0.0"
#
# data structures
@ -240,8 +240,8 @@ one-element range, but a range from tokenY to tokenY is the
full ring.
*/
struct KeyRange {
1: optional string start_key,
2: optional string end_key,
1: optional binary start_key,
2: optional binary end_key,
3: optional string start_token,
4: optional string end_token,
5: required i32 count=100
@ -255,7 +255,7 @@ struct KeyRange {
a SlicePredicate.
*/
struct KeySlice {
1: required string key,
1: required binary key,
2: required list<ColumnOrSuperColumn> columns,
}
@ -334,7 +334,7 @@ service Cassandra {
the only method that can throw an exception under non-failure conditions.)
*/
ColumnOrSuperColumn get(1:required string keyspace,
2:required string key,
2:required binary key,
3:required ColumnPath column_path,
4:required ConsistencyLevel consistency_level=ONE)
throws (1:InvalidRequestException ire, 2:NotFoundException nfe, 3:UnavailableException ue, 4:TimedOutException te),
@ -344,20 +344,20 @@ service Cassandra {
pair) specified by the given SlicePredicate. If no matching values are found, an empty list is returned.
*/
list<ColumnOrSuperColumn> get_slice(1:required string keyspace,
2:required string key,
2:required binary key,
3:required ColumnParent column_parent,
4:required SlicePredicate predicate,
5:required ConsistencyLevel consistency_level=ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
/**
Perform a get for column_path in parallel on the given list<string> keys. The return value maps keys to the
Perform a get for column_path in parallel on the given list<binary> keys. The return value maps keys to the
ColumnOrSuperColumn found. If no value corresponding to a key is present, the key will still be in the map, but both
the column and super_column references of the ColumnOrSuperColumn object it maps to will be null.
@deprecated; use multiget_slice
*/
map<string,ColumnOrSuperColumn> multiget(1:required string keyspace,
2:required list<string> keys,
map<binary,ColumnOrSuperColumn> multiget(1:required string keyspace,
2:required list<binary> keys,
3:required ColumnPath column_path,
4:required ConsistencyLevel consistency_level=ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
@ -365,8 +365,8 @@ service Cassandra {
/**
Performs a get_slice for column_parent and predicate for the given keys in parallel.
*/
map<string,list<ColumnOrSuperColumn>> multiget_slice(1:required string keyspace,
2:required list<string> keys,
map<binary,list<ColumnOrSuperColumn>> multiget_slice(1:required string keyspace,
2:required list<binary> keys,
3:required ColumnParent column_parent,
4:required SlicePredicate predicate,
5:required ConsistencyLevel consistency_level=ONE)
@ -376,7 +376,7 @@ service Cassandra {
returns the number of columns for a particular <code>key</code> and <code>ColumnFamily</code> or <code>SuperColumn</code>.
*/
i32 get_count(1:required string keyspace,
2:required string key,
2:required binary key,
3:required ColumnParent column_parent,
4:required ConsistencyLevel consistency_level=ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
@ -388,8 +388,8 @@ service Cassandra {
list<KeySlice> get_range_slice(1:required string keyspace,
2:required ColumnParent column_parent,
3:required SlicePredicate predicate,
4:required string start_key="",
5:required string finish_key="",
4:required binary start_key,
5:required binary finish_key,
6:required i32 row_count=100,
7:required ConsistencyLevel consistency_level=ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
@ -412,7 +412,7 @@ service Cassandra {
values -- it can only contain sub-Columns.
*/
void insert(1:required string keyspace,
2:required string key,
2:required binary key,
3:required ColumnPath column_path,
4:required binary value,
5:required i64 timestamp,
@ -426,7 +426,7 @@ service Cassandra {
@deprecated; use batch_mutate instead
*/
void batch_insert(1:required string keyspace,
2:required string key,
2:required binary key,
3:required map<string, list<ColumnOrSuperColumn>> cfmap,
4:required ConsistencyLevel consistency_level=ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
@ -437,7 +437,7 @@ service Cassandra {
row by just specifying the ColumnFamily, or you can remove a SuperColumn or a single Column by specifying those levels too.
*/
void remove(1:required string keyspace,
2:required string key,
2:required binary key,
3:required ColumnPath column_path,
4:required i64 timestamp,
5:ConsistencyLevel consistency_level=ONE)
@ -449,7 +449,7 @@ service Cassandra {
mutation_map maps key to column family to a list of Mutation objects to take place at that scope.
**/
void batch_mutate(1:required string keyspace,
2:required map<string, map<string, list<Mutation>>> mutation_map,
2:required map<binary, map<string, list<Mutation>>> mutation_map,
3:required ConsistencyLevel consistency_level=ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
@ -485,7 +485,7 @@ service Cassandra {
returns list of token strings such that first subrange is (list[0], list[1]],
next is (list[1], list[2]], etc. */
list<string> describe_splits(1:required string start_token,
2:required string end_token,
2:required string end_token,
3:required i32 keys_per_split),
void system_add_column_family(1:required CfDef cf_def)

445
interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
View File

File diff suppressed because it is too large Load Diff

2
interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java
View File

@ -42,6 +42,6 @@ import org.slf4j.LoggerFactory;
public class Constants {
public static final String VERSION = "3.0.0";
public static final String VERSION = "4.0.0";
}

48
interface/thrift/gen-java/org/apache/cassandra/thrift/KeyRange.java
View File

@ -61,8 +61,8 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
private static final TField END_TOKEN_FIELD_DESC = new TField("end_token", TType.STRING, (short)4);
private static final TField COUNT_FIELD_DESC = new TField("count", TType.I32, (short)5);
public String start_key;
public String end_key;
public byte[] start_key;
public byte[] end_key;
public String start_token;
public String end_token;
public int count;
@ -167,10 +167,12 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
__isset_bit_vector.clear();
__isset_bit_vector.or(other.__isset_bit_vector);
if (other.isSetStart_key()) {
this.start_key = other.start_key;
this.start_key = new byte[other.start_key.length];
System.arraycopy(other.start_key, 0, start_key, 0, other.start_key.length);
}
if (other.isSetEnd_key()) {
this.end_key = other.end_key;
this.end_key = new byte[other.end_key.length];
System.arraycopy(other.end_key, 0, end_key, 0, other.end_key.length);
}
if (other.isSetStart_token()) {
this.start_token = other.start_token;
@ -190,11 +192,11 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
return new KeyRange(this);
}
public String getStart_key() {
public byte[] getStart_key() {
return this.start_key;
}
public KeyRange setStart_key(String start_key) {
public KeyRange setStart_key(byte[] start_key) {
this.start_key = start_key;
return this;
}
@ -214,11 +216,11 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
}
}
public String getEnd_key() {
public byte[] getEnd_key() {
return this.end_key;
}
public KeyRange setEnd_key(String end_key) {
public KeyRange setEnd_key(byte[] end_key) {
this.end_key = end_key;
return this;
}
@ -315,7 +317,7 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
if (value == null) {
unsetStart_key();
} else {
setStart_key((String)value);
setStart_key((byte[])value);
}
break;
@ -323,7 +325,7 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
if (value == null) {
unsetEnd_key();
} else {
setEnd_key((String)value);
setEnd_key((byte[])value);
}
break;
@ -422,7 +424,7 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
if (this_present_start_key || that_present_start_key) {
if (!(this_present_start_key && that_present_start_key))
return false;
if (!this.start_key.equals(that.start_key))
if (!java.util.Arrays.equals(this.start_key, that.start_key))
return false;
}
@ -431,7 +433,7 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
if (this_present_end_key || that_present_end_key) {
if (!(this_present_end_key && that_present_end_key))
return false;
if (!this.end_key.equals(that.end_key))
if (!java.util.Arrays.equals(this.end_key, that.end_key))
return false;
}
@ -538,14 +540,14 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
switch (field.id) {
case 1: // START_KEY
if (field.type == TType.STRING) {
this.start_key = iprot.readString();
this.start_key = iprot.readBinary();
} else {
TProtocolUtil.skip(iprot, field.type);
}
break;
case 2: // END_KEY
if (field.type == TType.STRING) {
this.end_key = iprot.readString();
this.end_key = iprot.readBinary();
} else {
TProtocolUtil.skip(iprot, field.type);
}
@ -593,14 +595,14 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
if (this.start_key != null) {
if (isSetStart_key()) {
oprot.writeFieldBegin(START_KEY_FIELD_DESC);
oprot.writeString(this.start_key);
oprot.writeBinary(this.start_key);
oprot.writeFieldEnd();
}
}
if (this.end_key != null) {
if (isSetEnd_key()) {
oprot.writeFieldBegin(END_KEY_FIELD_DESC);
oprot.writeString(this.end_key);
oprot.writeBinary(this.end_key);
oprot.writeFieldEnd();
}
}
@ -635,7 +637,12 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
if (this.start_key == null) {
sb.append("null");
} else {
sb.append(this.start_key);
int __start_key_size = Math.min(this.start_key.length, 128);
for (int i = 0; i < __start_key_size; i++) {
if (i != 0) sb.append(" ");
sb.append(Integer.toHexString(this.start_key[i]).length() > 1 ? Integer.toHexString(this.start_key[i]).substring(Integer.toHexString(this.start_key[i]).length() - 2).toUpperCase() : "0" + Integer.toHexString(this.start_key[i]).toUpperCase());
}
if (this.start_key.length > 128) sb.append(" ...");
}
first = false;
}
@ -645,7 +652,12 @@ public class KeyRange implements TBase<KeyRange._Fields>, java.io.Serializable,
if (this.end_key == null) {
sb.append("null");
} else {
sb.append(this.end_key);
int __end_key_size = Math.min(this.end_key.length, 128);
for (int i = 0; i < __end_key_size; i++) {
if (i != 0) sb.append(" ");
sb.append(Integer.toHexString(this.end_key[i]).length() > 1 ? Integer.toHexString(this.end_key[i]).substring(Integer.toHexString(this.end_key[i]).length() - 2).toUpperCase() : "0" + Integer.toHexString(this.end_key[i]).toUpperCase());
}
if (this.end_key.length > 128) sb.append(" ...");
}
first = false;
}

26
interface/thrift/gen-java/org/apache/cassandra/thrift/KeySlice.java
View File

@ -57,7 +57,7 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
private static final TField KEY_FIELD_DESC = new TField("key", TType.STRING, (short)1);
private static final TField COLUMNS_FIELD_DESC = new TField("columns", TType.LIST, (short)2);
public String key;
public byte[] key;
public List<ColumnOrSuperColumn> columns;
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
@ -134,7 +134,7 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
}
public KeySlice(
String key,
byte[] key,
List<ColumnOrSuperColumn> columns)
{
this();
@ -147,7 +147,8 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
*/
public KeySlice(KeySlice other) {
if (other.isSetKey()) {
this.key = other.key;
this.key = new byte[other.key.length];
System.arraycopy(other.key, 0, key, 0, other.key.length);
}
if (other.isSetColumns()) {
List<ColumnOrSuperColumn> __this__columns = new ArrayList<ColumnOrSuperColumn>();
@ -167,11 +168,11 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
return new KeySlice(this);
}
public String getKey() {
public byte[] getKey() {
return this.key;
}
public KeySlice setKey(String key) {
public KeySlice setKey(byte[] key) {
this.key = key;
return this;
}
@ -236,7 +237,7 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
if (value == null) {
unsetKey();
} else {
setKey((String)value);
setKey((byte[])value);
}
break;
@ -304,7 +305,7 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
if (this_present_key || that_present_key) {
if (!(this_present_key && that_present_key))
return false;
if (!this.key.equals(that.key))
if (!java.util.Arrays.equals(this.key, that.key))
return false;
}
@ -366,7 +367,7 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
switch (field.id) {
case 1: // KEY
if (field.type == TType.STRING) {
this.key = iprot.readString();
this.key = iprot.readBinary();
} else {
TProtocolUtil.skip(iprot, field.type);
}
@ -406,7 +407,7 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
oprot.writeStructBegin(STRUCT_DESC);
if (this.key != null) {
oprot.writeFieldBegin(KEY_FIELD_DESC);
oprot.writeString(this.key);
oprot.writeBinary(this.key);
oprot.writeFieldEnd();
}
if (this.columns != null) {
@ -434,7 +435,12 @@ public class KeySlice implements TBase<KeySlice._Fields>, java.io.Serializable,
if (this.key == null) {
sb.append("null");
} else {
sb.append(this.key);
int __key_size = Math.min(this.key.length, 128);
for (int i = 0; i < __key_size; i++) {
if (i != 0) sb.append(" ");
sb.append(Integer.toHexString(this.key[i]).length() > 1 ? Integer.toHexString(this.key[i]).substring(Integer.toHexString(this.key[i]).length() - 2).toUpperCase() : "0" + Integer.toHexString(this.key[i]).toUpperCase());
}
if (this.key.length > 128) sb.append(" ...");
}
first = false;
if (!first) sb.append(", ");

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

@ -166,7 +166,7 @@ public class CliClient
colParent = new ColumnParent(columnFamily).setSuper_column(CliCompiler.getColumn(columnFamilySpec, 0).getBytes("UTF-8"));
}
int count = thriftClient_.get_count(tableName, key, colParent, ConsistencyLevel.ONE);
int count = thriftClient_.get_count(tableName, key.getBytes(), colParent, ConsistencyLevel.ONE);
css_.out.printf("%d columns\n", count);
}
@ -227,7 +227,7 @@ public class CliClient
columnName = CliCompiler.getColumn(columnFamilySpec, 1).getBytes("UTF-8");
}
thriftClient_.remove(tableName, key, new ColumnPath(columnFamily).setSuper_column(superColumnName).setColumn(columnName),
thriftClient_.remove(tableName, key.getBytes(), new ColumnPath(columnFamily).setSuper_column(superColumnName).setColumn(columnName),
timestampMicros(), ConsistencyLevel.ONE);
css_.out.println(String.format("%s removed.", (columnSpecCnt == 0) ? "row" : "column"));
}
@ -243,7 +243,7 @@ public class CliClient
throws InvalidRequestException, UnavailableException, TimedOutException, TException, UnsupportedEncodingException, IllegalAccessException, NotFoundException, InstantiationException, ClassNotFoundException
{
SliceRange range = new SliceRange(ArrayUtils.EMPTY_BYTE_ARRAY, ArrayUtils.EMPTY_BYTE_ARRAY, true, 1000000);
List<ColumnOrSuperColumn> columns = thriftClient_.get_slice(keyspace, key,
List<ColumnOrSuperColumn> columns = thriftClient_.get_slice(keyspace, key.getBytes(),
new ColumnParent(columnFamily).setSuper_column(superColumnName),
new SlicePredicate().setColumn_names(null).setSlice_range(range), ConsistencyLevel.ONE);
int size = columns.size();
@ -363,7 +363,7 @@ public class CliClient
// Perform a get(), print out the results.
ColumnPath path = new ColumnPath(columnFamily).setSuper_column(superColumnName).setColumn(columnName);
Column column = thriftClient_.get(tableName, key, path, ConsistencyLevel.ONE).column;
Column column = thriftClient_.get(tableName, key.getBytes(), path, ConsistencyLevel.ONE).column;
css_.out.printf("=> (column=%s, value=%s, timestamp=%d)\n", formatColumnName(tableName, columnFamily, column),
new String(column.value, "UTF-8"), column.timestamp);
}
@ -411,7 +411,7 @@ public class CliClient
}
// do the insert
thriftClient_.insert(tableName, key, new ColumnPath(columnFamily).setSuper_column(superColumnName).setColumn(columnName),
thriftClient_.insert(tableName, key.getBytes(), new ColumnPath(columnFamily).setSuper_column(superColumnName).setColumn(columnName),
value.getBytes(), timestampMicros(), ConsistencyLevel.ONE);
css_.out.println("Value inserted.");

4
src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java
View File

@ -60,7 +60,7 @@ import org.apache.thrift.transport.TTransportException;
*
* The default split size is 64k rows.
*/
public class ColumnFamilyInputFormat extends InputFormat<String, SortedMap<byte[], IColumn>>
public class ColumnFamilyInputFormat extends InputFormat<byte[], SortedMap<byte[], IColumn>>
{
private static final Logger logger = LoggerFactory.getLogger(StorageService.class);
@ -218,7 +218,7 @@ public class ColumnFamilyInputFormat extends InputFormat<String, SortedMap<byte[
}
@Override
public RecordReader<String, SortedMap<byte[], IColumn>> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException
public RecordReader<byte[], SortedMap<byte[], IColumn>> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException
{
return new ColumnFamilyRecordReader();
}

15
src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
View File

@ -46,11 +46,11 @@ import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransportException;
public class ColumnFamilyRecordReader extends RecordReader<String, SortedMap<byte[], IColumn>>
public class ColumnFamilyRecordReader extends RecordReader<byte[], SortedMap<byte[], IColumn>>
{
private ColumnFamilySplit split;
private RowIterator iter;
private Pair<String, SortedMap<byte[], IColumn>> currentRow;
private Pair<byte[], SortedMap<byte[], IColumn>> currentRow;
private SlicePredicate predicate;
private int totalRowCount; // total number of rows to fetch
private int batchRowCount; // fetch this many per batch
@ -59,7 +59,7 @@ public class ColumnFamilyRecordReader extends RecordReader<String, SortedMap<byt
public void close() {}
public String getCurrentKey()
public byte[] getCurrentKey()
{
return currentRow.left;
}
@ -95,7 +95,7 @@ public class ColumnFamilyRecordReader extends RecordReader<String, SortedMap<byt
return true;
}
private class RowIterator extends AbstractIterator<Pair<String, SortedMap<byte[], IColumn>>>
private class RowIterator extends AbstractIterator<Pair<byte[], SortedMap<byte[], IColumn>>>
{
private List<KeySlice> rows;
@ -159,8 +159,7 @@ public class ColumnFamilyRecordReader extends RecordReader<String, SortedMap<byt
// prepare for the next slice to be read
KeySlice lastRow = rows.get(rows.size() - 1);
IPartitioner p = DatabaseDescriptor.getPartitioner();
// FIXME: thrift strings
byte[] rowkey = lastRow.getKey().getBytes(UTF8);
byte[] rowkey = lastRow.getKey();
startToken = p.getTokenFactory().toString(p.getToken(rowkey));
}
catch (Exception e)
@ -213,7 +212,7 @@ public class ColumnFamilyRecordReader extends RecordReader<String, SortedMap<byt
}
@Override
protected Pair<String, SortedMap<byte[], IColumn>> computeNext()
protected Pair<byte[], SortedMap<byte[], IColumn>> computeNext()
{
maybeInit();
if (rows == null)
@ -227,7 +226,7 @@ public class ColumnFamilyRecordReader extends RecordReader<String, SortedMap<byt
IColumn column = unthriftify(cosc);
map.put(column.name(), column);
}
return new Pair<String, SortedMap<byte[], IColumn>>(ks.key, map);
return new Pair<byte[], SortedMap<byte[], IColumn>>(ks.key, map);
}
}

76
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -186,18 +186,17 @@ public class CassandraServer implements Cassandra.Iface
return thriftSuperColumns;
}
private Map<String, List<ColumnOrSuperColumn>> getSlice(List<ReadCommand> commands, ConsistencyLevel consistency_level)
private Map<byte[], List<ColumnOrSuperColumn>> getSlice(List<ReadCommand> commands, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
Map<byte[], ColumnFamily> columnFamilies = readColumnFamily(commands, consistency_level);
Map<String, List<ColumnOrSuperColumn>> columnFamiliesMap = new HashMap<String, List<ColumnOrSuperColumn>>();
Map<byte[], List<ColumnOrSuperColumn>> columnFamiliesMap = new HashMap<byte[], List<ColumnOrSuperColumn>>();
for (ReadCommand command: commands)
{
ColumnFamily cf = columnFamilies.get(command.key);
boolean reverseOrder = command instanceof SliceFromReadCommand && ((SliceFromReadCommand)command).reversed;
List<ColumnOrSuperColumn> thriftifiedColumns = thriftifyColumnFamily(cf, command.queryPath.superColumnName != null, reverseOrder);
// FIXME: string keys
columnFamiliesMap.put(new String(command.key, UTF8), thriftifiedColumns);
columnFamiliesMap.put(command.key, thriftifiedColumns);
}
return columnFamiliesMap;
@ -222,7 +221,7 @@ public class CassandraServer implements Cassandra.Iface
return thriftifyColumns(cf.getSortedColumns(), reverseOrder);
}
public List<ColumnOrSuperColumn> get_slice(String keyspace, String key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
public List<ColumnOrSuperColumn> get_slice(String keyspace, byte[] key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -232,7 +231,7 @@ public class CassandraServer implements Cassandra.Iface
return multigetSliceInternal(keyspace, Arrays.asList(key), column_parent, predicate, consistency_level).get(key);
}
public Map<String, List<ColumnOrSuperColumn>> multiget_slice(String keyspace, List<String> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
public Map<byte[], List<ColumnOrSuperColumn>> multiget_slice(String keyspace, List<byte[]> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -243,7 +242,7 @@ public class CassandraServer implements Cassandra.Iface
return multigetSliceInternal(keyspace, keys, column_parent, predicate, consistency_level);
}
private Map<String, List<ColumnOrSuperColumn>> multigetSliceInternal(String keyspace, List<String> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
private Map<byte[], List<ColumnOrSuperColumn>> multigetSliceInternal(String keyspace, List<byte[]> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
ThriftValidation.validateColumnParent(keyspace, column_parent);
@ -252,28 +251,26 @@ public class CassandraServer implements Cassandra.Iface
List<ReadCommand> commands = new ArrayList<ReadCommand>();
if (predicate.column_names != null)
{
for (String key: keys)
for (byte[] key: keys)
{
ThriftValidation.validateKey(key);
// FIXME: string keys
commands.add(new SliceByNamesReadCommand(keyspace, key.getBytes(UTF8), column_parent, predicate.column_names));
commands.add(new SliceByNamesReadCommand(keyspace, key, column_parent, predicate.column_names));
}
}
else
{
SliceRange range = predicate.slice_range;
for (String key: keys)
for (byte[] key: keys)
{
ThriftValidation.validateKey(key);
// FIXME: string keys
commands.add(new SliceFromReadCommand(keyspace, key.getBytes(UTF8), column_parent, range.start, range.finish, range.reversed, range.count));
commands.add(new SliceFromReadCommand(keyspace, key, column_parent, range.start, range.finish, range.reversed, range.count));
}
}
return getSlice(commands, consistency_level);
}
public ColumnOrSuperColumn get(String table, String key, ColumnPath column_path, ConsistencyLevel consistency_level)
public ColumnOrSuperColumn get(String table, byte[] key, ColumnPath column_path, ConsistencyLevel consistency_level)
throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -290,7 +287,7 @@ public class CassandraServer implements Cassandra.Iface
}
/** always returns a ColumnOrSuperColumn for each key, even if there is no data for it */
public Map<String, ColumnOrSuperColumn> multiget(String table, List<String> keys, ColumnPath column_path, ConsistencyLevel consistency_level)
public Map<byte[], ColumnOrSuperColumn> multiget(String table, List<byte[]> keys, ColumnPath column_path, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -301,7 +298,7 @@ public class CassandraServer implements Cassandra.Iface
return multigetInternal(table, keys, column_path, consistency_level);
}
private Map<String, ColumnOrSuperColumn> multigetInternal(String table, List<String> keys, ColumnPath column_path, ConsistencyLevel consistency_level)
private Map<byte[], ColumnOrSuperColumn> multigetInternal(String table, List<byte[]> keys, ColumnPath column_path, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
ThriftValidation.validateColumnPath(table, column_path);
@ -309,37 +306,34 @@ public class CassandraServer implements Cassandra.Iface
QueryPath path = new QueryPath(column_path.column_family, column_path.column == null ? null : column_path.super_column);
List<byte[]> nameAsList = Arrays.asList(column_path.column == null ? column_path.super_column : column_path.column);
List<ReadCommand> commands = new ArrayList<ReadCommand>();
for (String key: keys)
for (byte[] key: keys)
{
ThriftValidation.validateKey(key);
// FIXME: string keys
commands.add(new SliceByNamesReadCommand(table, key.getBytes(UTF8), path, nameAsList));
commands.add(new SliceByNamesReadCommand(table, key, path, nameAsList));
}
Map<String, ColumnOrSuperColumn> columnFamiliesMap = new HashMap<String, ColumnOrSuperColumn>();
Map<byte[], ColumnOrSuperColumn> columnFamiliesMap = new HashMap<byte[], ColumnOrSuperColumn>();
Map<byte[], ColumnFamily> cfamilies = readColumnFamily(commands, consistency_level);
for (ReadCommand command: commands)
{
ColumnFamily cf = cfamilies.get(command.key);
// FIXME: string keys
String skey = new String(command.key, UTF8);
if (cf == null)
{
columnFamiliesMap.put(skey, new ColumnOrSuperColumn());
columnFamiliesMap.put(command.key, new ColumnOrSuperColumn());
}
else
{
List<ColumnOrSuperColumn> tcolumns = thriftifyColumnFamily(cf, command.queryPath.superColumnName != null, false);
columnFamiliesMap.put(skey, tcolumns.size() > 0 ? tcolumns.iterator().next() : new ColumnOrSuperColumn());
columnFamiliesMap.put(command.key, tcolumns.size() > 0 ? tcolumns.iterator().next() : new ColumnOrSuperColumn());
}
}
return columnFamiliesMap;
}
public int get_count(String table, String key, ColumnParent column_parent, ConsistencyLevel consistency_level)
public int get_count(String table, byte[] key, ColumnParent column_parent, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -352,7 +346,7 @@ public class CassandraServer implements Cassandra.Iface
return get_slice(table, key, column_parent, predicate, consistency_level).size();
}
public void insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, ConsistencyLevel consistency_level)
public void insert(String table, byte[] key, ColumnPath column_path, byte[] value, long timestamp, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -363,8 +357,7 @@ public class CassandraServer implements Cassandra.Iface
ThriftValidation.validateKey(key);
ThriftValidation.validateColumnPath(table, column_path);
// FIXME: string keys
RowMutation rm = new RowMutation(table, key.getBytes(UTF8));
RowMutation rm = new RowMutation(table, key);
try
{
rm.add(new QueryPath(column_path), value, timestamp);
@ -376,7 +369,7 @@ public class CassandraServer implements Cassandra.Iface
doInsert(consistency_level, rm);
}
public void batch_insert(String keyspace, String key, Map<String, List<ColumnOrSuperColumn>> cfmap, ConsistencyLevel consistency_level)
public void batch_insert(String keyspace, byte[] key, Map<String, List<ColumnOrSuperColumn>> cfmap, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -394,11 +387,10 @@ public class CassandraServer implements Cassandra.Iface
}
}
// FIXME: string keys
doInsert(consistency_level, RowMutation.getRowMutation(keyspace, key.getBytes(UTF8), cfmap));
doInsert(consistency_level, RowMutation.getRowMutation(keyspace, key, cfmap));
}
public void batch_mutate(String keyspace, Map<String,Map<String,List<Mutation>>> mutation_map, ConsistencyLevel consistency_level)
public void batch_mutate(String keyspace, Map<byte[],Map<String,List<Mutation>>> mutation_map, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -425,9 +417,9 @@ public class CassandraServer implements Cassandra.Iface
checkLoginAuthorized(needed);
List<RowMutation> rowMutations = new ArrayList<RowMutation>();
for (Map.Entry<String, Map<String, List<Mutation>>> mutationEntry: mutation_map.entrySet())
for (Map.Entry<byte[], Map<String, List<Mutation>>> mutationEntry: mutation_map.entrySet())
{
String key = mutationEntry.getKey();
byte[] key = mutationEntry.getKey();
ThriftValidation.validateKey(key);
Map<String, List<Mutation>> columnFamilyToMutations = mutationEntry.getValue();
@ -440,8 +432,7 @@ public class CassandraServer implements Cassandra.Iface
ThriftValidation.validateMutation(keyspace, cfName, mutation);
}
}
// FIXME: string keys
rowMutations.add(RowMutation.getRowMutationFromMutations(keyspace, key.getBytes(UTF8), columnFamilyToMutations));
rowMutations.add(RowMutation.getRowMutationFromMutations(keyspace, key, columnFamilyToMutations));
}
if (consistency_level == ConsistencyLevel.ZERO)
{
@ -460,7 +451,7 @@ public class CassandraServer implements Cassandra.Iface
}
}
public void remove(String table, String key, ColumnPath column_path, long timestamp, ConsistencyLevel consistency_level)
public void remove(String table, byte[] key, ColumnPath column_path, long timestamp, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@ -471,8 +462,7 @@ public class CassandraServer implements Cassandra.Iface
ThriftValidation.validateKey(key);
ThriftValidation.validateColumnPathOrParent(table, column_path);
// FIXME: string keys
RowMutation rm = new RowMutation(table, key.getBytes(UTF8));
RowMutation rm = new RowMutation(table, key);
rm.delete(new QueryPath(column_path), timestamp);
doInsert(consistency_level, rm);
@ -523,7 +513,7 @@ public class CassandraServer implements Cassandra.Iface
return columnFamiliesMap;
}
public List<KeySlice> get_range_slice(String keyspace, ColumnParent column_parent, SlicePredicate predicate, String start_key, String finish_key, int maxRows, ConsistencyLevel consistency_level)
public List<KeySlice> get_range_slice(String keyspace, ColumnParent column_parent, SlicePredicate predicate, byte[] start_key, byte[] finish_key, int maxRows, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TException, TimedOutException
{
if (logger.isDebugEnabled())
@ -565,8 +555,7 @@ public class CassandraServer implements Cassandra.Iface
}
else
{
// FIXME: string keys
bounds = new Bounds(p.getToken(range.start_key.getBytes(UTF8)), p.getToken(range.end_key.getBytes(UTF8)));
bounds = new Bounds(p.getToken(range.start_key), p.getToken(range.end_key));
}
rows = StorageProxy.getRangeSlice(new RangeSliceCommand(keyspace, column_parent, predicate, bounds, range.count), consistency_level);
assert rows != null;
@ -585,8 +574,7 @@ public class CassandraServer implements Cassandra.Iface
for (Row row : rows)
{
List<ColumnOrSuperColumn> thriftifiedColumns = thriftifyColumnFamily(row.cf, column_parent.super_column != null, reversed);
// FIXME: string keys
keySlices.add(new KeySlice(new String(row.key.key, UTF8), thriftifiedColumns));
keySlices.add(new KeySlice(row.key.key, thriftifiedColumns));
}
return keySlices;

19
src/java/org/apache/cassandra/thrift/ThriftValidation.java
View File

@ -38,16 +38,18 @@ import org.apache.cassandra.utils.FBUtilities;
public class ThriftValidation
{
static void validateKey(String key) throws InvalidRequestException
static void validateKey(byte[] key) throws InvalidRequestException
{
if (key.isEmpty())
if (key == null || key.length == 0)
{
throw new InvalidRequestException("Key may not be empty");
}
// check that writeUTF will be able to handle it -- encoded length must fit in 2 bytes
int utflen = FBUtilities.encodedUTF8Length(key);
if (utflen > 65535)
throw new InvalidRequestException("Encoded key length of " + utflen + " is longer than maximum of 65535");
// check that key can be handled by FBUtilities.writeShortByteArray
if (key.length > FBUtilities.MAX_UNSIGNED_SHORT)
{
throw new InvalidRequestException("Key length of " + key.length +
" is longer than maximum of " + FBUtilities.MAX_UNSIGNED_SHORT);
}
}
private static void validateTable(String tablename) throws KeyspaceNotDefinedException
@ -303,9 +305,8 @@ public class ThriftValidation
if (range.start_key != null)
{
IPartitioner p = StorageService.getPartitioner();
// FIXME: string keys
Token startToken = p.getToken(range.start_key.getBytes(FBUtilities.UTF8));
Token endToken = p.getToken(range.end_key.getBytes(FBUtilities.UTF8));
Token startToken = p.getToken(range.start_key);
Token endToken = p.getToken(range.end_key);
if (startToken.compareTo(endToken) > 0 && !endToken.equals(p.getMinimumToken()))
{
if (p instanceof RandomPartitioner)

8
test/unit/org/apache/cassandra/client/TestRingCache.java
View File

@ -87,20 +87,20 @@ public class TestRingCache
for (int nRows = minRow; nRows < maxRow; nRows++)
{
String row = rowPrefix + nRows;
byte[] row = (rowPrefix + nRows).getBytes();
ColumnPath col = new ColumnPath("Standard1").setSuper_column(null).setColumn("col1".getBytes());
List<InetAddress> endPoints = tester.ringCache.getEndPoint(row.getBytes());
List<InetAddress> endPoints = tester.ringCache.getEndPoint(row);
String hosts="";
for (int i = 0; i < endPoints.size(); i++)
hosts = hosts + ((i > 0) ? "," : "") + endPoints.get(i);
System.out.println("hosts with key " + row + " : " + hosts + "; choose " + endPoints.get(0));
System.out.println("hosts with key " + new String(row) + " : " + hosts + "; choose " + endPoints.get(0));
// now, read the row back directly from the host owning the row locally
tester.setup(endPoints.get(0).getHostAddress(), DatabaseDescriptor.getRpcPort());
tester.thriftClient.insert(keyspace, row, col, "val1".getBytes(), 1, ConsistencyLevel.ONE);
Column column = tester.thriftClient.get(keyspace, row, col, ConsistencyLevel.ONE).column;
System.out.println("read row " + row + " " + new String(column.name) + ":" + new String(column.value) + ":" + column.timestamp);
System.out.println("read row " + new String(row) + " " + new String(column.name) + ":" + new String(column.value) + ":" + column.timestamp);
}
System.exit(1);

2
test/unit/org/apache/cassandra/service/EmbeddedCassandraServiceTest.java
View File

@ -89,7 +89,7 @@ public class EmbeddedCassandraServiceTest
{
Cassandra.Client client = getClient();
String key_user_id = "1";
byte[] key_user_id = "1".getBytes();
long timestamp = System.currentTimeMillis();
ColumnPath cp = new ColumnPath("Standard1");