Minor simplification and sanity checks in UnsafeSorter

jeanlyn · May 1, 2015 · 3db12de · 3db12de
1 parent 767d3ca
commit 3db12de
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 6 deletions.
diff --git a/.gitignore b/.gitignore
@@ -29,6 +29,7 @@ conf/*.properties
 conf/*.conf
 conf/*.xml
 conf/slaves
+core/build/py4j/
 docs/_site
 docs/api
 target/

diff --git a/core/src/main/java/org/apache/spark/unsafe/sort/UnsafeSorter.java b/core/src/main/java/org/apache/spark/unsafe/sort/UnsafeSorter.java
@@ -30,12 +30,22 @@ public static final class KeyPointerAndPrefix {
      * A pointer to a record; see {@link org.apache.spark.unsafe.memory.TaskMemoryManager} for a
      * description of how these addresses are encoded.
      */
-    long recordPointer;
+    public long recordPointer;
 
     /**
      * A key prefix, for use in comparisons.
      */
-    long keyPrefix;
+    public long keyPrefix;
+
+    @Override
+    public int hashCode() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      throw new UnsupportedOperationException();
+    }
   }
 
   public static abstract class RecordComparator {
@@ -115,8 +125,9 @@ public void insertRecord(long objectAddress) {
     final long baseOffset = memoryManager.getOffsetInPage(objectAddress);
     final long keyPrefix = prefixComputer.computePrefix(baseObject, baseOffset);
     sortBuffer[sortBufferInsertPosition] = objectAddress;
-    sortBuffer[sortBufferInsertPosition + 1] = keyPrefix;
-    sortBufferInsertPosition += 2;
+    sortBufferInsertPosition++;
+    sortBuffer[sortBufferInsertPosition] = keyPrefix;
+    sortBufferInsertPosition++;
   }
 
   public Iterator<KeyPointerAndPrefix> getSortedIterator() {

diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -313,7 +313,8 @@ object SparkEnv extends Logging {
     // Let the user specify short names for shuffle managers
     val shortShuffleMgrNames = Map(
       "hash" -> "org.apache.spark.shuffle.hash.HashShuffleManager",
-      "sort" -> "org.apache.spark.shuffle.sort.SortShuffleManager")
+      "sort" -> "org.apache.spark.shuffle.sort.SortShuffleManager",
+      "unsafe" -> "org.apache.spark.shuffle.unsafe.UnsafeShuffleManager")
     val shuffleMgrName = conf.get("spark.shuffle.manager", "sort")
     val shuffleMgrClass = shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase, shuffleMgrName)
     val shuffleManager = instantiateClass[ShuffleManager](shuffleMgrClass)

diff --git a/core/src/main/scala/org/apache/spark/shuffle/FileShuffleBlockManager.scala b/core/src/main/scala/org/apache/spark/shuffle/FileShuffleBlockManager.scala
@@ -34,7 +34,7 @@ import org.apache.spark.storage._
 import org.apache.spark.util.{MetadataCleaner, MetadataCleanerType, TimeStampedHashMap}
 import org.apache.spark.util.collection.{PrimitiveKeyOpenHashMap, PrimitiveVector}
 
-/** A group of writers for a ShuffleMapTask, one writer per reducer. */
+/** A group of writers for ShuffleMapTask, one writer per reducer. */
 private[spark] trait ShuffleWriterGroup {
   val writers: Array[BlockObjectWriter]