Merge pull request #484 from nevillelyh/develop

add immutable version of sorted(Reverse)Take

Author: ianoc

algebird-core/src/main/scala/com/twitter/algebird/Aggregator.scala

@@ -102,6 +102,16 @@ object Aggregator extends java.io.Serializable {
    */
   def sortedReverseTake[T: Ordering](count: Int): MonoidAggregator[T, PriorityQueue[T], Seq[T]] =
     new mutable.PriorityQueueToListAggregator[T](count)(implicitly[Ordering[T]].reverse)
+  /**
+   * Immutable version of sortedTake, for frameworks that check immutability of reduce functions.
+   */
+  def immutableSortedTake[T: Ordering](count: Int): MonoidAggregator[T, TopK[T], Seq[T]] =
+    new TopKToListAggregator[T](count)
+  /**
+   * Immutable version of sortedReverseTake, for frameworks that check immutability of reduce functions.
+   */
+  def immutableSortedReverseTake[T: Ordering](count: Int): MonoidAggregator[T, TopK[T], Seq[T]] =
+    new TopKToListAggregator[T](count)(implicitly[Ordering[T]].reverse)
   /**
    * Put everything in a List. Note, this could fill the memory if the List is very large.
    */

algebird-core/src/main/scala/com/twitter/algebird/TopKMonoid.scala

@@ -87,3 +87,9 @@ class TopKMonoid[T](k: Int)(implicit ord: Ordering[T]) extends Monoid[TopK[T]] {
     TopK(size, reversed, max)
   }
 }
+
+class TopKToListAggregator[A](max: Int)(implicit ord: Ordering[A]) extends MonoidAggregator[A, TopK[A], List[A]] {
+  val monoid: Monoid[TopK[A]] = new TopKMonoid[A](max)(ord)
+  override def present(a: TopK[A]): List[A] = a.items
+  override def prepare(a: A): TopK[A] = TopK(1, List(a), Some(a))
+}