Initial import of code from Databricks unsafe utils repo.

Author: JoshRosen

pom.xml

@@ -97,6 +97,7 @@
     <module>sql/catalyst</module>
     <module>sql/core</module>
     <module>sql/hive</module>
+    <module>unsafe</module>
     <module>assembly</module>
     <module>external/twitter</module>
     <module>external/flume</module>

project/SparkBuild.scala

@@ -34,11 +34,11 @@ object BuildCommons {
 
   val allProjects@Seq(bagel, catalyst, core, graphx, hive, hiveThriftServer, mllib, repl,
     sql, networkCommon, networkShuffle, streaming, streamingFlumeSink, streamingFlume, streamingKafka,
-    streamingMqtt, streamingTwitter, streamingZeromq, launcher) =
+    streamingMqtt, streamingTwitter, streamingZeromq, launcher, unsafe) =
     Seq("bagel", "catalyst", "core", "graphx", "hive", "hive-thriftserver", "mllib", "repl",
       "sql", "network-common", "network-shuffle", "streaming", "streaming-flume-sink",
       "streaming-flume", "streaming-kafka", "streaming-mqtt", "streaming-twitter",
-      "streaming-zeromq", "launcher").map(ProjectRef(buildLocation, _))
+      "streaming-zeromq", "launcher", "unsafe").map(ProjectRef(buildLocation, _))
 
   val optionallyEnabledProjects@Seq(yarn, yarnStable, java8Tests, sparkGangliaLgpl,
     sparkKinesisAsl) = Seq("yarn", "yarn-stable", "java8-tests", "ganglia-lgpl",

unsafe/src/main/java/org/apache/spark/unsafe/PlatformDependent.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.unsafe;
+
+import java.lang.reflect.Field;
+
+import sun.misc.Unsafe;
+
+public final class PlatformDependent {
+
+  public static final Unsafe UNSAFE;
+
+  public static final int BYTE_ARRAY_OFFSET;
+
+  public static final int INT_ARRAY_OFFSET;
+
+  public static final int LONG_ARRAY_OFFSET;
+
+  public static final int DOUBLE_ARRAY_OFFSET;
+
+  /**
+   * Limits the number of bytes to copy per {@link Unsafe#copyMemory(long, long, long)} to
+   * allow safepoint polling during a large copy.
+   */
+  private static final long UNSAFE_COPY_THRESHOLD = 1024L * 1024L;
+
+  static {
+    sun.misc.Unsafe unsafe;
+    try {
+      Field unsafeField = Unsafe.class.getDeclaredField("theUnsafe");
+      unsafeField.setAccessible(true);
+      unsafe = (sun.misc.Unsafe) unsafeField.get(null);
+    } catch (Throwable cause) {
+      unsafe = null;
+    }
+    UNSAFE = unsafe;
+
+    if (UNSAFE != null) {
+      BYTE_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
+      INT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(int[].class);
+      LONG_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(long[].class);
+      DOUBLE_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(double[].class);
+    } else {
+      BYTE_ARRAY_OFFSET = 0;
+      INT_ARRAY_OFFSET = 0;
+      LONG_ARRAY_OFFSET = 0;
+      DOUBLE_ARRAY_OFFSET = 0;
+    }
+  }
+
+  static public void copyMemory(
+      Object src,
+      long srcOffset,
+      Object dst,
+      long dstOffset,
+      long length) {
+    while (length > 0) {
+      long size = Math.min(length, UNSAFE_COPY_THRESHOLD);
+      UNSAFE.copyMemory(src, srcOffset, dst, dstOffset, size);
+      length -= size;
+      srcOffset += size;
+      dstOffset += size;
+    }
+  }
+
+  /**
+   * Raises an exception bypassing compiler checks for checked exceptions.
+   */
+  public static void throwException(Throwable t) {
+    UNSAFE.throwException(t);
+  }
+}

unsafe/src/main/java/org/apache/spark/unsafe/array/ByteArrayMethods.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.unsafe.array;
+
+import org.apache.spark.unsafe.PlatformDependent;
+
+import java.lang.Object;
+
+public class ByteArrayMethods {
+
+  private ByteArrayMethods() {
+    // Private constructor, since this class only contains static methods.
+  }
+
+  /**
+   * Optimized byte array equality check for 8-byte-word-aligned byte arrays.
+   * @return true if the arrays are equal, false otherwise
+   */
+  public static boolean wordAlignedArrayEquals(
+      Object leftBaseObject,
+      long leftBaseOffset,
+      Object rightBaseObject,
+      long rightBaseOffset,
+      long arrayLengthInBytes) {
+    for (int i = 0; i < arrayLengthInBytes; i += 8) {
+      final long left =
+        PlatformDependent.UNSAFE.getLong(leftBaseObject, leftBaseOffset + i);
+      final long right =
+        PlatformDependent.UNSAFE.getLong(rightBaseObject, rightBaseOffset + i);
+      if (left != right) return false;
+    }
+    return true;
+  }
+}

unsafe/src/main/java/org/apache/spark/unsafe/array/LongArray.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.unsafe.array;
+
+import org.apache.spark.unsafe.PlatformDependent;
+import org.apache.spark.unsafe.memory.MemoryBlock;
+
+/**
+ * An array of long values. Compared with native JVM arrays, this:
+ * <ul>
+ *   <li>supports using both in-heap and off-heap memory</li>
+ *   <li>supports 64-bit addressing, i.e. array length greater than {@code Integer.MAX_VALUE}</li>
+ *   <li>has no bound checking, and thus can crash the JVM process when assert is turned off</li>
+ * </ul>
+ */
+public final class LongArray {
+
+  private static final int WIDTH = 8;
+  private static final long ARRAY_OFFSET = PlatformDependent.LONG_ARRAY_OFFSET;
+
+  private final MemoryBlock memory;
+  private final Object baseObj;
+  private final long baseOffset;
+
+  private final long length;
+
+  public LongArray(MemoryBlock memory) {
+    assert memory.size() % WIDTH == 0 : "Memory not aligned (" + memory.size() + ")";
+    this.memory = memory;
+    this.baseObj = memory.getBaseObject();
+    this.baseOffset = memory.getBaseOffset();
+    this.length = memory.size() / WIDTH;
+  }
+
+  public MemoryBlock memoryBlock() {
+    return memory;
+  }
+
+  /**
+   * Returns the number of elements this array can hold.
+   */
+  public long size() {
+    return length;
+  }
+
+  /**
+   * Sets the value at position {@code index}.
+   */
+  public void set(long index, long value) {
+    assert index >= 0 : "index (" + index + ") should >= 0";
+    assert index < length : "index (" + index + ") should < length (" + length + ")";
+    PlatformDependent.UNSAFE.putLong(baseObj, baseOffset + index * WIDTH, value);
+  }
+
+  /**
+   * Returns the value at position {@code index}.
+   */
+  public long get(long index) {
+    assert index >= 0 : "index (" + index + ") should >= 0";
+    assert index < length : "index (" + index + ") should < length (" + length + ")";
+    return PlatformDependent.UNSAFE.getLong(baseObj, baseOffset + index * WIDTH);
+  }
+
+  /**
+   * Returns a copy of the array as a JVM native array. The caller should make sure this array's
+   * length is less than {@code Integer.MAX_VALUE}.
+   */
+  public long[] toJvmArray() throws IndexOutOfBoundsException {
+    if (length > Integer.MAX_VALUE) {
+      throw new IndexOutOfBoundsException(
+        "array size (" + length + ") too large and cannot be converted into JVM array");
+    }
+
+    final long[] arr = new long[(int) length];
+    PlatformDependent.UNSAFE.copyMemory(
+      baseObj,
+      baseOffset,
+      arr,
+      ARRAY_OFFSET,
+      length * WIDTH);
+    return arr;
+  }
+}

unsafe/src/main/java/org/apache/spark/unsafe/bitset/BitSet.java

@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.unsafe.bitset;
+
+import org.apache.spark.unsafe.array.LongArray;
+import org.apache.spark.unsafe.memory.MemoryBlock;
+
+/**
+ * A fixed size uncompressed bit set backed by a {@link LongArray}.
+ *
+ * Each bit occupies exactly one bit of storage.
+ */
+public final class BitSet {
+
+  /** A long array for the bits. */
+  private final LongArray words;
+
+  /** Length of the long array. */
+  private final long numWords;
+
+  /**
+   * Creates a new {@link BitSet} using the specified memory block. Size of the memory block must be
+   * multiple of 8 bytes (i.e. 64 bits).
+   */
+  public BitSet(MemoryBlock memory) {
+    words = new LongArray(memory);
+    numWords = words.size();
+  }
+
+  public MemoryBlock memoryBlock() {
+    return words.memoryBlock();
+  }
+
+  /**
+   * Returns the number of bits in this {@code BitSet}.
+   */
+  public long capacity() {
+    return numWords * 64;
+  }
+
+  /**
+   * Sets the bit at the specified index to {@code true}.
+   */
+  public void set(long index) {
+    assert index < numWords * 64 : "index (" + index + ") should < length (" + numWords * 64 + ")";
+    BitSetMethods.set(
+      words.memoryBlock().getBaseObject(), words.memoryBlock().getBaseOffset(), index);
+  }
+
+  /**
+   * Sets the bit at the specified index to {@code false}.
+   */
+  public void unset(long index) {
+    assert index < numWords * 64 : "index (" + index + ") should < length (" + numWords * 64 + ")";
+    BitSetMethods.unset(
+      words.memoryBlock().getBaseObject(), words.memoryBlock().getBaseOffset(), index);
+  }
+
+  /**
+   * Returns {@code true} if the bit is set at the specified index.
+   */
+  public boolean isSet(long index) {
+    assert index < numWords * 64 : "index (" + index + ") should < length (" + numWords * 64 + ")";
+    return BitSetMethods.isSet(
+      words.memoryBlock().getBaseObject(), words.memoryBlock().getBaseOffset(), index);
+  }
+
+  /**
+   * Returns the number of bits set to {@code true} in this {@link BitSet}.
+   */
+  public long cardinality() {
+    long sum = 0L;
+    for (long i = 0; i < numWords; i++) {
+      sum += java.lang.Long.bitCount(words.get(i));
+    }
+    return sum;
+  }
+
+  /**
+   * Returns the index of the first bit that is set to true that occurs on or after the
+   * specified starting index. If no such bit exists then {@code -1} is returned.
+   * <p>
+   * To iterate over the true bits in a BitSet, use the following loop:
+   * <pre>
+   * <code>
+   *  for (long i = bs.nextSetBit(0); i >= 0; i = bs.nextSetBit(i + 1)) {
+   *    // operate on index i here
+   *  }
+   * </code>
+   * </pre>
+   *
+   * @param fromIndex the index to start checking from (inclusive)
+   * @return the index of the next set bit, or -1 if there is no such bit
+   */
+  public long nextSetBit(long fromIndex) {
+    return BitSetMethods.nextSetBit(
+      words.memoryBlock().getBaseObject(),
+      words.memoryBlock().getBaseOffset(),
+      fromIndex,
+      numWords);
+  }
+}