EventuallyAggregator and variants

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

@@ -168,3 +168,38 @@ class EventuallyRing[E, O](convert: O => E)(mustConvert: O => Boolean)(implicit
   }
 
 }
+
+trait AbstractEventuallyAggregator[A, E, O, C]
+  extends Aggregator[A, Either[E, O], C] {
+  def prepare(a: A) = Right(rightAggregator.prepare(a))
+  def present(b: Either[E, O]) = b match {
+    case Right(o) => rightAggregator.present(o)
+    case Left(e) => presentLeft(e)
+  }
+
+  def presentLeft(e: E): C
+
+  def convert(o: O): E
+  def mustConvert(o: O): Boolean
+
+  def leftSemigroup: Semigroup[E]
+  def rightAggregator: Aggregator[A, O, C]
+}
+
+trait EventuallyAggregator[A, E, O, C]
+  extends AbstractEventuallyAggregator[A, E, O, C] {
+
+  //avoid init order issues and cyclical references
+  @transient lazy val semigroup =
+    new EventuallySemigroup[E, O](convert)(mustConvert)(leftSemigroup, rightAggregator.semigroup)
+}
+
+trait EventuallyMonoidAggregator[A, E, O, C]
+  extends AbstractEventuallyAggregator[A, E, O, C]
+  with MonoidAggregator[A, Either[E, O], C] {
+
+  def rightAggregator: MonoidAggregator[A, O, C]
+
+  @transient lazy val monoid =
+    new EventuallyMonoid[E, O](convert)(mustConvert)(leftSemigroup, rightAggregator.monoid)
+}

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

@@ -451,3 +451,15 @@ case class HyperLogLogAggregator(val hllMonoid: HyperLogLogMonoid) extends Monoi
   def prepare(value: Array[Byte]) = monoid.create(value)
   def present(hll: HLL) = hll
 }
+
+case class SetSizeAggregator[A](hllBits: Int, maxSetSize: Int = 10)(implicit toBytes: A => Array[Byte])
+  extends EventuallyMonoidAggregator[A, HLL, Set[A], Long] {
+
+  def presentLeft(hll: HLL) = hll.approximateSize.estimate
+
+  def mustConvert(set: Set[A]) = set.size > maxSetSize
+  def convert(set: Set[A]) = leftSemigroup.batchCreate(set.map(toBytes))
+
+  val leftSemigroup = new HyperLogLogMonoid(hllBits)
+  val rightAggregator = Aggregator.uniqueCount[A].andThenPresent { _.toLong }
+}