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

fix SlicePredicate serialization inside Hadoop jobs. patch by jbellis; reviewed by eevans for CASSANDRA-1049 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.6@947771 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-05-24 19:45:48 +00:00
parent a8ce7f002d
commit 2d683c0a7c
5 changed files with 51 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -26,6 +26,7 @@
* expose PhiConvictThreshold (CASSANDRA-1053)
* make repair of RF==1 a no-op (CASSANDRA-1090)
* improve default JVM GC options (CASSANDRA-1014)
* fix SlicePredicate serialization inside Hadoop jobs (CASSANDRA-1049)
0.6.1

3
ivy.xml
View File

@ -20,6 +20,9 @@
<info organisation="apache-cassandra" module="cassandra"/>
<dependencies>
<!-- for Hadoop unit tests -->
<dependency org="commons-logging" name="commons-logging" rev="1.1.1"/>
<!-- FIXME: paranamer can be dropped after we're depending on avro
(since it depends on them). -->
<dependency org="com.thoughtworks.paranamer"

11
src/java/org/apache/cassandra/hadoop/ConfigHelper.java
View File

@ -24,11 +24,12 @@ package org.apache.cassandra.hadoop;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.ThriftValidation;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.hadoop.conf.Configuration;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TJSONProtocol;
import org.apache.thrift.protocol.TBinaryProtocol;
public class ConfigHelper
{
@ -135,10 +136,10 @@ public class ConfigHelper
{
assert predicate != null;
// this is so awful it's kind of cool!
TSerializer serializer = new TSerializer(new TJSONProtocol.Factory());
TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
try
{
return serializer.toString(predicate, "UTF-8");
return FBUtilities.bytesToHex(serializer.serialize(predicate));
}
catch (TException e)
{
@ -149,11 +150,11 @@ public class ConfigHelper
private static SlicePredicate predicateFromString(String st)
{
assert st != null;
TDeserializer deserializer = new TDeserializer(new TJSONProtocol.Factory());
TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
SlicePredicate predicate = new SlicePredicate();
try
{
deserializer.deserialize(predicate, st, "UTF-8");
deserializer.deserialize(predicate, FBUtilities.hexToBytes(st));
}
catch (TException e)
{

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

@ -22,6 +22,7 @@ import java.io.*;
import java.math.BigInteger;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
@ -38,7 +39,6 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.thrift.TBase;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
@ -480,4 +480,11 @@ public class FBUtilities
}
return utflen;
}
public static byte[] toByteArray(long n)
{
byte[] bytes = new byte[8];
ByteBuffer.wrap(bytes).putLong(n);
return bytes;
}
}

33
test/unit/org/apache/cassandra/hadoop/ColumnFamilyInputFormatTest.java Normal file
View File

@ -0,0 +1,33 @@
package org.apache.cassandra.hadoop;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.junit.Test;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.hadoop.conf.Configuration;
public class ColumnFamilyInputFormatTest
{
@Test
public void testSlicePredicate()
{
long columnValue = 1271253600000l;
byte[] columnBytes = FBUtilities.toByteArray(columnValue);
List<byte[]> columnNames = new ArrayList<byte[]>();
columnNames.add(columnBytes);
SlicePredicate originalPredicate = new SlicePredicate().setColumn_names(columnNames);
Configuration conf = new Configuration();
ConfigHelper.setSlicePredicate(conf, originalPredicate);
SlicePredicate rtPredicate = ConfigHelper.getSlicePredicate(conf);
assert rtPredicate.column_names.size() == 1;
assert Arrays.equals(originalPredicate.column_names.get(0), rtPredicate.column_names.get(0));
}
}