[SPARK-8301] [SQL] Improve UTF8String substring/startsWith/endsWith/contains performance

Jira: https://issues.apache.org/jira/browse/SPARK-8301

Added the private method startsWith(prefix, offset) to implement startsWith, endsWith and contains without copying the array

I hope that the component SQL is still correct. I copied it from the Jira ticket.

Author: Tarek Auel <tarek.auel@googlemail.com>
Author: Tarek Auel <tarek.auel@gmail.com>

Closes #6804 from tarekauel/SPARK-8301 and squashes the following commits:

f5d6b9a [Tarek Auel] fixed parentheses and annotation
6d7b068 [Tarek Auel] [SPARK-8301] removed null checks
9ca0473 [Tarek Auel] [SPARK-8301] removed null checks
1c327eb [Tarek Auel] [SPARK-8301] removed new
9f17cc8 [Tarek Auel] [SPARK-8301] fixed conversion byte to string in codegen
3a0040f [Tarek Auel] [SPARK-8301] changed call of UTF8String.set to UTF8String.from
e4530d2 [Tarek Auel] [SPARK-8301] changed call of UTF8String.set to UTF8String.from
a5f853a [Tarek Auel] [SPARK-8301] changed visibility of set to protected. Changed annotation of bytes from Nullable to Nonnull
d2fb05f [Tarek Auel] [SPARK-8301] added additional null checks
79cb55b [Tarek Auel] [SPARK-8301] null check. Added test cases for null check.
b17909e [Tarek Auel] [SPARK-8301] removed unnecessary copying of UTF8String. Added a private function startsWith(prefix, offset) to implement the check for startsWith, endsWith and contains.

1 files changed, 18 insertions, 12 deletions

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 a351680195..9871a70a40 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
@@ -20,7 +20,7 @@ package org.apache.spark.unsafe.types;
 import java.io.Serializable;
 import java.io.UnsupportedEncodingException;
 import java.util.Arrays;
-import javax.annotation.Nullable;
+import javax.annotation.Nonnull;
 
 import org.apache.spark.unsafe.PlatformDependent;
 
@@ -34,7 +34,7 @@ import org.apache.spark.unsafe.PlatformDependent;
  */
 public final class UTF8String implements Comparable<UTF8String>, Serializable {
 
-  @Nullable
+  @Nonnull
   private byte[] bytes;
 
   private static int[] bytesOfCodePointInUTF8 = {2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2,
@@ -55,7 +55,7 @@ public final class UTF8String implements Comparable<UTF8String>, Serializable {
   /**
    * Updates the UTF8String with String.
    */
-  public UTF8String set(final String str) {
+  protected UTF8String set(final String str) {
     try {
       bytes = str.getBytes("utf-8");
     } catch (UnsupportedEncodingException e) {
@@ -69,7 +69,7 @@ public final class UTF8String implements Comparable<UTF8String>, Serializable {
   /**
    * Updates the UTF8String with byte[], which should be encoded in UTF-8.
    */
-  public UTF8String set(final byte[] bytes) {
+  protected UTF8String set(final byte[] bytes) {
     this.bytes = bytes;
     return this;
   }
@@ -131,24 +131,30 @@ public final class UTF8String implements Comparable<UTF8String>, Serializable {
     }
 
     for (int i = 0; i <= bytes.length - b.length; i++) {
-      // TODO: Avoid copying.
-      if (bytes[i] == b[0] && Arrays.equals(Arrays.copyOfRange(bytes, i, i + b.length), b)) {
+      if (bytes[i] == b[0] && startsWith(b, i)) {
         return true;
       }
     }
     return false;
   }
 
+  private boolean startsWith(final byte[] prefix, int offsetInBytes) {
+    if (prefix.length + offsetInBytes > bytes.length || offsetInBytes < 0) {
+      return false;
+    }
+    int i = 0;
+    while (i < prefix.length && prefix[i] == bytes[i + offsetInBytes]) {
+      i++;
+    }
+    return i == prefix.length;
+  }
+
   public boolean startsWith(final UTF8String prefix) {
-    final byte[] b = prefix.getBytes();
-    // TODO: Avoid copying.
-    return b.length <= bytes.length && Arrays.equals(Arrays.copyOfRange(bytes, 0, b.length), b);
+    return startsWith(prefix.getBytes(), 0);
   }
 
   public boolean endsWith(final UTF8String suffix) {
-    final byte[] b = suffix.getBytes();
-    return b.length <= bytes.length &&
-      Arrays.equals(Arrays.copyOfRange(bytes, bytes.length - b.length, bytes.length), b);
+    return startsWith(suffix.getBytes(), bytes.length - suffix.getBytes().length);
   }
 
   public UTF8String toUpperCase() {