From 2d683c0a7c9c2ab878c521369a007d12c58df677 Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Mon, 24 May 2010 19:45:48 +0000
Subject: [PATCH] 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
---
 CHANGES.txt                                   |  1 +
 ivy.xml                                       |  3 ++
 .../apache/cassandra/hadoop/ConfigHelper.java | 11 ++++---
 .../apache/cassandra/utils/FBUtilities.java   |  9 ++++-
 .../hadoop/ColumnFamilyInputFormatTest.java   | 33 +++++++++++++++++++
 5 files changed, 51 insertions(+), 6 deletions(-)
 create mode 100644 test/unit/org/apache/cassandra/hadoop/ColumnFamilyInputFormatTest.java

diff --git a/CHANGES.txt b/CHANGES.txt
index 1702b070dd..091e37f650 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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
diff --git a/ivy.xml b/ivy.xml
index f40f630186..b5d8bbd528 100644
--- a/ivy.xml
+++ b/ivy.xml
@@ -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"
diff --git a/src/java/org/apache/cassandra/hadoop/ConfigHelper.java b/src/java/org/apache/cassandra/hadoop/ConfigHelper.java
index a3cc97658d..284afd3b0d 100644
--- a/src/java/org/apache/cassandra/hadoop/ConfigHelper.java
+++ b/src/java/org/apache/cassandra/hadoop/ConfigHelper.java
@@ -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)
         {
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index f2d9abd467..1280c4c7c4 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -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;
+    }
 }
diff --git a/test/unit/org/apache/cassandra/hadoop/ColumnFamilyInputFormatTest.java b/test/unit/org/apache/cassandra/hadoop/ColumnFamilyInputFormatTest.java
new file mode 100644
index 0000000000..068fc529af
--- /dev/null
+++ b/test/unit/org/apache/cassandra/hadoop/ColumnFamilyInputFormatTest.java
@@ -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));
+    }
+}