From d40a2511cdc193cc8771fb0164ebdf9ce8473c97 Mon Sep 17 00:00:00 2001
From: Davies Liu <davies@databricks.com>
Date: Fri, 13 Nov 2015 13:10:20 -0800
Subject: [PATCH] support Kryo for UTF8String

---
 unsafe/pom.xml                                |  4 ++++
 .../apache/spark/unsafe/types/UTF8String.java | 24 +++++++++++++++++--
 2 files changed, 26 insertions(+), 2 deletions(-)

diff --git a/unsafe/pom.xml b/unsafe/pom.xml
index caf1f77890b5..a1c1111364ee 100644
--- a/unsafe/pom.xml
+++ b/unsafe/pom.xml
@@ -36,6 +36,10 @@
   </properties>
 
   <dependencies>
+    <dependency>
+      <groupId>com.twitter</groupId>
+      <artifactId>chill_${scala.binary.version}</artifactId>
+    </dependency>
 
     <!-- Core dependencies -->
     <dependency>
diff --git a/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java b/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java
index b7aecb5102ba..4bd3fd777207 100644
--- a/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java
+++ b/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java
@@ -24,6 +24,11 @@
 import java.util.Arrays;
 import java.util.Map;
 
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.KryoSerializable;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
 import org.apache.spark.unsafe.Platform;
 import org.apache.spark.unsafe.array.ByteArrayMethods;
 
@@ -38,9 +43,9 @@
  * <p>
  * Note: This is not designed for general use cases, should not be used outside SQL.
  */
-public final class UTF8String implements Comparable<UTF8String>, Externalizable {
+public final class UTF8String implements Comparable<UTF8String>, Externalizable, KryoSerializable {
 
-  // These are only updated by readExternal()
+  // These are only updated by readExternal() or read()
   @Nonnull
   private Object base;
   private long offset;
@@ -1003,4 +1008,19 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
     in.readFully((byte[]) base);
   }
 
+  @Override
+  public void write(Kryo kryo, Output out) {
+    byte[] bytes = getBytes();
+    out.writeInt(bytes.length);
+    out.write(bytes);
+  }
+
+  @Override
+  public void read(Kryo kryo, Input in) {
+    this.offset = BYTE_ARRAY_OFFSET;
+    this.numBytes = in.readInt();
+    this.base = new byte[numBytes];
+    in.read((byte[]) base);
+  }
+
 }