Add sorted join

[johnynek]

Something like:

val left: TypedPipe[(K, V1)]
val right TypedPipe[(K, V2)]

// assumes all the Right come first
def split[V1, V2](it: Iterator[Either[V1, V2]]): (Iterator[V1], List[V2]) = {
  val buf = it.buffered

  @annotation.tailrec
  def go(rights: List[V2]): List[V2] =
    if (buf.isEmpty) rights
    else {
      buf.head match {
        case Right(v2) => buf.next; go(v2 :: rights)
        case Left(_) => rights
      }
    }
  (buf, go(Nil).reverse)
}

left.eitherValues(right)
  .group
  .sort
  .mapValueStream { it: Iterator[Either[V1, V2]] =>
    // now you have all the V2 first then the V1 in sorted order
    val (lefts, rights) = split(it)
    for {
      l <- lefts
      r <- rights.iterator
    } yield (l, r)
  }

We can even imagine not using cascading at all joins. Supporting nested sorted joins will be a bit of work.