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

update contrib WordCount, ClientOnlyExample for Thrift 0.5. patch by jbellis 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1026377 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-10-22 15:31:27 +00:00
parent e2409452b5
commit 2d9b4ba3cc
9 changed files with 59 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
NEWS.txt
View File

@ -62,6 +62,9 @@ Thrift API
----------
- The Cassandra server now defaults to framed mode, rather than
unframed. Unframed is obsolete and will be removed in the future.
- The Cassandra Thrift interface file has been updated for Thrift 0.5.
If you are compiling your own client code from the interface, you
will need to upgrade the Thrift compiler to match.
- Row keys are now bytes: keys stored by versions prior to 0.7.0 will be
returned as UTF-8 encoded bytes. OrderPreservingPartitioner and
CollatingOrderPreservingPartitioner continue to expect that keys contain

34
contrib/client_only/ClientOnlyExample.java
View File

@ -16,21 +16,19 @@
* limitations under the License.
*/
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.service.*;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.UnavailableException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeoutException;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.utils.ByteBufferUtil;
public class ClientOnlyExample
{
@ -49,12 +47,11 @@ public class ClientOnlyExample
}
// do some writing.
final AbstractType comp = ColumnFamily.getComparatorFor("Keyspace1", "Standard1", null);
for (int i = 0; i < 100; i++)
{
RowMutation change = new RowMutation("Keyspace1", ("key" + i).getBytes());
RowMutation change = new RowMutation("Keyspace1", ByteBuffer.wrap(("key" + i).getBytes()));
ColumnPath cp = new ColumnPath("Standard1").setColumn(("colb").getBytes());
change.add(new QueryPath(cp), ("value" + i).getBytes(), 0);
change.add(new QueryPath(cp), ByteBuffer.wrap(("value" + i).getBytes()), 0);
// don't call change.apply(). The reason is that is makes a static call into Table, which will perform
// local storage initialization, which creates local directories.
@ -81,14 +78,15 @@ public class ClientOnlyExample
}
// do some queries.
Collection<byte[]> cols = new ArrayList<byte[]>()
Collection<ByteBuffer> cols = new ArrayList<ByteBuffer>()
{{
add("colb".getBytes());
add(ByteBuffer.wrap("colb".getBytes()));
}};
for (int i = 0; i < 100; i++)
{
List<ReadCommand> commands = new ArrayList<ReadCommand>();
SliceByNamesReadCommand readCommand = new SliceByNamesReadCommand("Keyspace1", ("key" + i).getBytes(), new QueryPath("Standard1", null, null), cols);
SliceByNamesReadCommand readCommand = new SliceByNamesReadCommand("Keyspace1", ByteBuffer.wrap(("key" + i).getBytes()),
new QueryPath("Standard1", null, null), cols);
readCommand.setDigestQuery(false);
commands.add(readCommand);
List<Row> rows = StorageProxy.readProtocol(commands, ConsistencyLevel.ONE);
@ -99,7 +97,7 @@ public class ClientOnlyExample
{
for (IColumn col : cf.getSortedColumns())
{
System.out.println(new String(col.name()) + ", " + new String(col.value()));
System.out.println(ByteBufferUtil.string(col.name(), Charsets.UTF_8) + ", " + ByteBufferUtil.string(col.value(), Charsets.UTF_8));
}
}
else

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

@ -31,6 +31,7 @@ import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.hadoop.ColumnFamilyInputFormat;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
@ -80,7 +81,7 @@ public class WordCount extends Configured implements Tool
IColumn column = columns.get(columnName.getBytes());
if (column == null)
return;
String value = new String(column.value());
String value = ByteBufferUtil.string(column.value(), Charsets.UTF_8);
logger.debug("read " + key + ":" + value + " from " + context.getInputSplit());
StringTokenizer itr = new StringTokenizer(value);
@ -214,11 +215,9 @@ public class WordCount extends Configured implements Tool
ConfigHelper.setInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setPartitioner(job.getConfiguration(), "org.apache.cassandra.dht.RandomPartitioner");
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(columnName.getBytes()));
SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(ByteBuffer.wrap(columnName.getBytes())));
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
job.waitForCompletion(true);
}
return 0;

5
src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
View File

@ -23,6 +23,8 @@ import java.io.*;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import com.google.common.base.Charsets;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
@ -30,6 +32,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.net.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import static com.google.common.base.Charsets.UTF_8;
@ -60,7 +63,7 @@ public class RowMutationVerbHandler implements IVerbHandler
{
ByteBuffer addressBytes = FBUtilities.readShortByteArray(dis);
if (logger_.isDebugEnabled())
logger_.debug("Adding hint for " + InetAddress.getByName(new String(addressBytes.array(),addressBytes.position()+addressBytes.arrayOffset(),addressBytes.remaining())));
logger_.debug("Adding hint for " + InetAddress.getByName(ByteBufferUtil.string(addressBytes, Charsets.UTF_8)));
RowMutation hintedMutation = new RowMutation(Table.SYSTEM_TABLE, addressBytes);
hintedMutation.addHints(rm);
hintedMutation.apply();

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

@ -40,6 +40,7 @@ import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
@ -251,13 +252,11 @@ public class SystemTable
IColumn clusterCol = cf.getColumn(CLUSTERNAME);
assert partitionerCol != null;
assert clusterCol != null;
if (!DatabaseDescriptor.getPartitioner().getClass().getName().equals(
new String(partitionerCol.value().array(),
partitionerCol.value().position()+partitionerCol.value().arrayOffset(),
partitionerCol.value().remaining(), UTF_8)))
if (!DatabaseDescriptor.getPartitioner().getClass().getName().equals(ByteBufferUtil.string(partitionerCol.value(), UTF_8)))
throw new ConfigurationException("Detected partitioner mismatch! Did you change the partitioner?");
if (!DatabaseDescriptor.getClusterName().equals(new String(clusterCol.value().array(),clusterCol.value().position()+clusterCol.value().arrayOffset(),clusterCol.value().remaining())))
throw new ConfigurationException("Saved cluster name " + new String(clusterCol.value().array(),clusterCol.value().position()+clusterCol.value().arrayOffset(),clusterCol.value().remaining()) + " != configured name " + DatabaseDescriptor.getClusterName());
String savedClusterName = ByteBufferUtil.string(clusterCol.value(), UTF_8);
if (!DatabaseDescriptor.getClusterName().equals(savedClusterName))
throw new ConfigurationException("Saved cluster name " + savedClusterName + " != configured name " + DatabaseDescriptor.getClusterName());
}
public static Token getSavedToken()

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

@ -24,6 +24,10 @@ package org.apache.cassandra.db.marshal;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import com.google.common.base.Charsets;
import org.apache.cassandra.utils.ByteBufferUtil;
public class AsciiType extends BytesType
{
public static final AsciiType instance = new AsciiType();
@ -33,13 +37,6 @@ public class AsciiType extends BytesType
@Override
public String getString(ByteBuffer bytes)
{
try
{
return new String(bytes.array(),bytes.position()+bytes.arrayOffset(),bytes.remaining(), "US-ASCII");
}
catch (UnsupportedEncodingException e)
{
throw new RuntimeException(e);
}
return ByteBufferUtil.string(bytes, Charsets.US_ASCII);
}
}

24
src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java
View File

@ -24,7 +24,10 @@ import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.Random;
import com.google.common.base.Charsets;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
@ -109,29 +112,16 @@ public class OrderPreservingPartitioner implements IPartitioner<StringToken>
return new StringToken(buffer.toString());
}
private final Token.TokenFactory<String> tokenFactory = new Token.TokenFactory<String>() {
private final Token.TokenFactory<String> tokenFactory = new Token.TokenFactory<String>()
{
public ByteBuffer toByteArray(Token<String> stringToken)
{
try
{
return ByteBuffer.wrap(stringToken.token.getBytes("UTF-8"));
}
catch (UnsupportedEncodingException e)
{
throw new RuntimeException(e);
}
return ByteBuffer.wrap(stringToken.token.getBytes(Charsets.UTF_8));
}
public Token<String> fromByteArray(ByteBuffer bytes)
{
try
{
return new StringToken(new String(bytes.array(),bytes.position()+bytes.arrayOffset(),bytes.limit(), "UTF-8"));
}
catch (UnsupportedEncodingException e)
{
throw new RuntimeException(e);
}
return new StringToken(ByteBufferUtil.string(bytes, Charsets.UTF_8));
}
public String toString(Token<String> stringToken)

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

@ -109,6 +109,7 @@ import org.apache.cassandra.streaming.StreamRequestVerbHandler;
import org.apache.cassandra.streaming.StreamingService;
import org.apache.cassandra.thrift.Constants;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.SkipNullRepresenter;
import org.apache.cassandra.utils.WrappedRunnable;
@ -121,6 +122,7 @@ import org.yaml.snakeyaml.DumperOptions;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.nodes.Tag;
import com.google.common.base.Charsets;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
@ -2031,7 +2033,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
RawColumnDefinition rcd = new RawColumnDefinition();
rcd.index_name = cd.index_name;
rcd.index_type = cd.index_type;
rcd.name = new String(cd.name.array(),cd.name.position()+cd.name.arrayOffset(),cd.name.remaining(), "UTF8");
rcd.name = ByteBufferUtil.string(cd.name, Charsets.UTF_8);
rcd.validator_class = cd.validator.getClass().getName();
rcf.column_metadata[j++] = rcd;
}

13
src/java/org/apache/cassandra/utils/ByteBufferUtil.java
View File

@ -18,7 +18,9 @@
*/
package org.apache.cassandra.utils;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
/**
* Utility methods to make ByteBuffers less painful
@ -35,8 +37,19 @@ public class ByteBufferUtil {
{
return FBUtilities.compareUnsigned(o1, o2.array(), 0, o2.arrayOffset()+o2.position(), o1.length, o2.limit());
}
public static int compare(ByteBuffer o1, byte[] o2)
{
return FBUtilities.compareUnsigned(o1.array(), o2, o1.arrayOffset()+o1.position(), 0, o1.limit(), o2.length);
}
public static String string(ByteBuffer b, Charset charset)
{
return new String(b.array(), b.arrayOffset() + b.position(), b.remaining(), charset);
}
public static String string(ByteBuffer b)
{
return new String(b.array(), b.arrayOffset() + b.position(), b.remaining());
}
}