Merge pull request apache#503 from pwendell/master

Fix bug on read-side of external sort when using Snappy.

This case wasn't handled correctly and this patch fixes it.

core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala

@@ -358,7 +358,15 @@ private[spark] class ExternalAppendOnlyMap[K, V, C](
   private class DiskMapIterator(file: File, blockId: BlockId) extends Iterator[(K, C)] {
     val fileStream = new FileInputStream(file)
     val bufferedStream = new FastBufferedInputStream(fileStream, fileBufferSize)
-    val compressedStream = blockManager.wrapForCompression(blockId, bufferedStream)
+
+    val shouldCompress = blockManager.shouldCompress(blockId)
+    val compressionCodec = new LZFCompressionCodec(sparkConf)
+    val compressedStream =
+      if (shouldCompress) {
+        compressionCodec.compressedInputStream(bufferedStream)
+      } else {
+        bufferedStream
+      }
     var deserializeStream = ser.deserializeStream(compressedStream)
     var objectsRead = 0