Merge pull request twitter#389 from DanielleSucher/aggregator-zip

Add Aggregator.zip

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

@@ -224,6 +224,22 @@ trait Aggregator[-A, B, +C] extends java.io.Serializable { self =>
   def join[A2 <: A, B2, C2](that: Aggregator[A2, B2, C2]): Aggregator[A2, (B, B2), (C, C2)] =
     GeneratedTupleAggregator.from2((this, that))
 
+  /**
+   * This allows you to join two aggregators into one that takes a tuple input,
+   * which in turn allows you to chain .composePrepare onto the result if you have
+   * an initial input that has to be prepared differently for each of the joined aggregators.
+   *
+   * The law here is: ag1.zip(ag2).apply(as.zip(bs)) == (ag1(as), ag2(bs))
+   */
+  def zip[A2, B2, C2](ag2: Aggregator[A2, B2, C2]): Aggregator[(A, A2), (B, B2), (C, C2)] = {
+    val ag1 = this
+    new Aggregator[(A, A2), (B, B2), (C, C2)] {
+      def prepare(a: (A, A2)) = (ag1.prepare(a._1), ag2.prepare(a._2))
+      val semigroup = new Tuple2Semigroup()(ag1.semigroup, ag2.semigroup)
+      def present(b: (B, B2)) = (ag1.present(b._1), ag2.present(b._2))
+    }
+  }
+
   /**
    * An Aggregator can be converted to a Fold, but not vice-versa
    * Note, a Fold is more constrained so only do this if you require

algebird-test/src/test/scala/com/twitter/algebird/AggregatorLaws.scala

@@ -54,6 +54,7 @@ class AggregatorLaws extends PropSpec with PropertyChecks with Matchers {
       assert(in.isEmpty || c(in) == (ag1(in), ag2(in)))
     }
   }
+
   property("Applicative composing two Aggregators is correct") {
     forAll { (in: List[Int], ag1: Aggregator[Int, Set[Int], Int], ag2: Aggregator[Int, Unit, String]) =>
       type AggInt[T] = Aggregator[Int, _, T]
@@ -62,6 +63,14 @@ class AggregatorLaws extends PropSpec with PropertyChecks with Matchers {
     }
   }
 
+  property("Aggregator.zip composing two Aggregators is correct") {
+    forAll { (in: List[(Int, String)], ag1: Aggregator[Int, Int, Int], ag2: Aggregator[String, Set[String], Double]) =>
+      val c = ag1.zip(ag2)
+      val (as, bs) = in.unzip
+      assert(in.isEmpty || c(in) == (ag1(as), ag2(bs)))
+    }
+  }
+
   property("Aggregator.lift works for empty sequences") {
     forAll { (in: List[Int], ag: Aggregator[Int, Int, Int]) =>
       val liftedAg = ag.lift