[WIP] PullByteBufferOut as a default ordering

scalding-serialization/src/main/scala/com/twitter/scalding/serialization/Exported.scala

@@ -0,0 +1,20 @@
+/*
+Licensed 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 com.twitter.scalding.serialization
+
+// We wrap types in Exported to provide low priority implicits
+// the real type has a low priority implicit to extract from Exported
+// into the original type.
+// See more @ https://github.com/milessabin/export-hook
+case class Exported[T](instance: T) extends AnyVal

scalding-serialization/src/main/scala/com/twitter/scalding/serialization/OrderedSerialization.scala

@@ -32,9 +32,17 @@ trait OrderedSerialization[T] extends Ordering[T] with Serialization[T] {
    * the InputStreams is mutated to be the end of the record.
    */
   def compareBinary(a: InputStream, b: InputStream): OrderedSerialization.Result
+
+  /**
+   * This compares two InputStreams. After this call, the position in
+   * the InputStreams may or may not be at the end of the record.
+   */
+  def compareBinaryNoConsume(a: InputStream, b: InputStream): OrderedSerialization.Result = {
+    compareBinary(a, b)
+  }
 }
 
-object OrderedSerialization {
+object OrderedSerialization extends LowPriorityOrderedSerialization {
   /**
    * Represents the result of a comparison that might fail due
    * to an error deserializing
@@ -214,3 +222,8 @@ final case class DeserializingOrderedSerialization[T](serialization: Serializati
   final override def staticSize = serialization.staticSize
   final override def dynamicSize(t: T) = serialization.dynamicSize(t)
 }
+
+private[serialization] trait LowPriorityOrderedSerialization {
+  implicit final def importedOrderedSerialization[A](implicit exported: Exported[OrderedSerialization[A]]): OrderedSerialization[A] = exported.instance
+}
+

scalding-serialization/src/main/scala/com/twitter/scalding/serialization/Serialization.scala

@@ -38,9 +38,10 @@ import scala.util.hashing.Hashing
  * implementation. This must satisfy:
  *   (!equiv(a, b)) || (hash(a) == hash(b))
  */
-trait Serialization[T] extends Equiv[T] with Hashing[T] with Serializable {
+trait Serialization[T] extends Equiv[T] with Hashing[T] with Serializable with LowPrioritySerialization {
   def read(in: InputStream): Try[T]
   def write(out: OutputStream, t: T): Try[Unit]
+
   /**
    * If all items have a static size, this returns Some, else None
    * NOTE: lawful implementations that return Some here much return
@@ -53,6 +54,11 @@ trait Serialization[T] extends Equiv[T] with Hashing[T] with Serializable {
    * otherwise the caller should just serialize into an ByteArrayOutputStream
    */
   def dynamicSize(t: T): Option[Int]
+
+  // Override this to provide more efficient
+  def skip(in: InputStream): Try[Unit] = {
+    read(in).map{ _ => () }
+  }
 }
 
 /**
@@ -171,3 +177,7 @@ object Serialization {
       sizeLaw,
       transitivity)
 }
+
+private[serialization] trait LowPrioritySerialization {
+  implicit final def importedSerialization[A](implicit exported: Exported[Serialization[A]]): Serialization[A] = exported.instance
+}

scalding-serialization/src/main/scala/com/twitter/scalding/serialization/macros/impl/OrderedBufferableProviderImpl.scala

@@ -41,7 +41,6 @@ object OrderedSerializationProviderImpl {
     val stringDispatcher = StringOrderedBuf.dispatch(c)
     val traversablesDispatcher = TraversablesOrderedBuf.dispatch(c)(buildDispatcher)
     val unitDispatcher = UnitOrderedBuf.dispatch(c)
-    val byteBufferDispatcher = ByteBufferOrderedBuf.dispatch(c)
     val sealedTraitDispatcher = SealedTraitOrderedBuf.dispatch(c)(buildDispatcher)
 
     OrderedSerializationProviderImpl
@@ -51,7 +50,6 @@ object OrderedSerializationProviderImpl {
       .orElse(optionDispatcher)
       .orElse(eitherDispatcher)
       .orElse(stringDispatcher)
-      .orElse(byteBufferDispatcher)
       .orElse(traversablesDispatcher)
       .orElse(caseClassDispatcher)
       .orElse(caseObjectDispatcher)

scalding-serialization/src/main/scala/com/twitter/scalding/serialization/provided/OrderedSerializationByteBuffer.scala

@@ -0,0 +1,71 @@
+package com.twitter.scalding.serialization.provided
+
+import com.twitter.scalding.serialization.JavaStreamEnrichments._
+import java.nio.ByteBuffer
+import java.io.InputStream
+import com.twitter.scalding.serialization.Serialization
+import com.twitter.scalding.serialization.JavaStreamEnrichments._
+import scala.math.min
+import scala.util.{ Failure, Try }
+import scala.util.control.NonFatal
+
+object OrderedSerializationByteBuffer extends UnsafeOrderedSerialization[ByteBuffer] {
+
+  override def staticSize: Option[Int] = None
+
+  def hash(x: ByteBuffer): Int =
+    x.hashCode
+
+  override def unsafeSwitchingCompareBinaryNoConsume(inputStreamA: InputStream, inputStreamB: InputStream, consumeToEnd: Boolean): Int = {
+    val lenA = inputStreamA.readPosVarInt
+    val lenB = inputStreamB.readPosVarInt
+    val queryLength = scala.math.min(lenA, lenB)
+    var incr = 0
+    var state = 0
+
+    while (incr < queryLength && state == 0) {
+      state = java.lang.Byte.compare(inputStreamA.readByte, inputStreamB.readByte)
+      incr = incr + 1
+    }
+    if (consumeToEnd) {
+      inputStreamA.skip(lenA - incr)
+      inputStreamB.skip(lenB - incr)
+    }
+    if (state == 0) {
+      java.lang.Integer.compare(lenA, lenB)
+    } else {
+      state
+    }
+  }
+
+  override def skip(inputStream: InputStream): Try[Unit] = {
+    try {
+      val lenA = inputStream.readPosVarInt
+      inputStream.skip(lenA)
+      Serialization.successUnit
+    } catch {
+      case NonFatal(e) =>
+        Failure(e)
+    }
+  }
+
+  def unsafeWrite(outputStream: java.io.OutputStream, element: ByteBuffer): Unit = {
+    outputStream.writePosVarInt(element.remaining)
+    outputStream
+      .writeBytes(element.array, element.arrayOffset + element.position, element.remaining)
+  }
+
+  def unsafeRead(inputStream: java.io.InputStream): ByteBuffer = {
+    val lenA = inputStream.readPosVarInt
+    val bytes = new Array[Byte](lenA)
+    inputStream.readFully(bytes)
+    java.nio.ByteBuffer.wrap(bytes)
+  }
+
+  def compare(a: ByteBuffer, b: ByteBuffer): Int = a.compareTo(b)
+
+  def dynamicSize(element: ByteBuffer): Option[Int] = Some {
+    val tmpLen = element.remaining
+    posVarIntSize(tmpLen) + tmpLen
+  }
+}

scalding-serialization/src/main/scala/com/twitter/scalding/serialization/provided/UnsafeCompareBinary.scala

@@ -0,0 +1,74 @@
+/*
+Copyright 2015 Twitter, Inc.
+
+Licensed 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 com.twitter.scalding.serialization.provided
+
+import com.twitter.scalding.serialization.JavaStreamEnrichments._
+
+import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, InputStream, OutputStream }
+import scala.util.{ Failure, Success, Try }
+import scala.util.control.NonFatal
+import com.twitter.scalding.serialization._
+
+abstract class UnsafeOrderedSerialization[T] extends OrderedSerialization[T] {
+  // This will write out the interior data as a blob with no prepended length
+  // This means binary compare cannot skip on this data.
+  // However the contract remains that one should be able to _read_ the data
+  // back out again.
+  def unsafeWrite(out: java.io.OutputStream, t: T): Unit
+  // This is an internal read method that matches the unsafe write
+  // importantly any outer length header added to enable skipping isn't included here
+  def unsafeRead(in: java.io.InputStream): T
+
+  // This is an inner binary compare that the user must supply
+  def unsafeSwitchingCompareBinaryNoConsume(inputStreamA: InputStream, inputStreamB: InputStream, shouldConsume: Boolean): Int
+
+  // This is the public write method, if we need to inject a size head of the object
+  // this is where we do it!
+  final override def write(into: OutputStream, e: T): Try[Unit] =
+    try {
+      unsafeWrite(into, e)
+      Serialization.successUnit
+    } catch {
+      case NonFatal(e) =>
+        Failure(e)
+    }
+
+  final def read(in: InputStream): Try[T] =
+    try {
+      Success(unsafeRead(in))
+    } catch {
+      case NonFatal(e) =>
+        Failure(e)
+    }
+
+  override def compareBinaryNoConsume(inputStreamA: InputStream, inputStreamB: InputStream): OrderedSerialization.Result =
+    try {
+      val r = unsafeSwitchingCompareBinaryNoConsume(inputStreamA, inputStreamB, false)
+      OrderedSerialization.resultFrom(r)
+    } catch {
+      case NonFatal(e) =>
+        OrderedSerialization.CompareFailure(e)
+    }
+
+  override def compareBinary(inputStreamA: InputStream, inputStreamB: InputStream): OrderedSerialization.Result =
+    try {
+      val r = unsafeSwitchingCompareBinaryNoConsume(inputStreamA, inputStreamB, true)
+      OrderedSerialization.resultFrom(r)
+    } catch {
+      case NonFatal(e) =>
+        OrderedSerialization.CompareFailure(e)
+    }
+}

scalding-serialization/src/main/scala/com/twitter/scalding/serialization/provided/package.scala

@@ -0,0 +1,9 @@
+package com.twitter.scalding.serialization
+
+import java.nio.ByteBuffer
+
+import com.twitter.scalding.serialization.provided.{ OrderedSerializationByteBuffer }
+
+package object provided {
+  implicit val byteBufferOrderedSerialization: Exported[OrderedSerialization[ByteBuffer]] = Exported(OrderedSerializationByteBuffer)
+}