Add SummingWithHitsCache class to also track key hits.

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

@@ -33,7 +33,7 @@ object SummingCache {
 /**
  * A Stateful Summer on Map[K,V] that keeps a cache of recent keys
  */
-class SummingCache[K, V] private (capacity: Int)(implicit sgv: Semigroup[V])
+class SummingCache[K, V](capacity: Int)(implicit sgv: Semigroup[V])
   extends StatefulSummer[Map[K, V]] {
 
   require(capacity >= 0, "Cannot have negative capacity in SummingIterator")
@@ -76,3 +76,18 @@ class SummingCache[K, V] private (capacity: Int)(implicit sgv: Semigroup[V])
       }
   }).asScala
 }
+
+object SummingWithHitsCache {
+  def apply[K, V: Semigroup](cap: Int): SummingWithHitsCache[K, V] = new SummingWithHitsCache[K, V](cap)
+}
+/**
+ * A SummingCache that also tracks the number of key hits
+ */
+class SummingWithHitsCache[K, V] (capacity: Int)(implicit sgv: Semigroup[V])
+  extends SummingCache[K, V](capacity)(sgv) {
+
+  def putWithHits(m: Map[K, V]): (Int, Option[Map[K, V]]) = {
+    val keyHits = m.keys.count(cache.contains)
+    (keyHits, put(m))
+  }
+}

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

@@ -53,6 +53,51 @@ class SummingCacheTest extends PropSpec with PropertyChecks with Matchers {
   }
 }
 
+class SummingWithHitsCacheTest extends SummingCacheTest {
+  import SummingCacheTest._
+
+  val RAND = new scala.util.Random
+
+  def getHits[K, V: Monoid](c: Capacity, items: List[(K, V)]) = {
+    val sc = SummingWithHitsCache[K, V](c.cap)
+    val mitems = items.map { Map(_) }
+    mitems.map { sc.putWithHits(_)._1 }.tail
+  }
+
+  property("hit rates will always be 1 for stream with the same key") {
+    forAll { (c: Capacity, values: List[Int]) =>
+      // Only run this when we have at least 2 items and non-zero cap
+      if (values.size > 1 && c.cap > 1) {
+        val key = RAND.nextInt
+        val items = values.map { (key, _) }
+        val keyHits = getHits(c, items)
+        assert(keyHits.filter{ _ != 1 }.size == 0)
+      }
+    }
+  }
+
+  property("hit rates will always be 0 when cap is 0") {
+    forAll { items: List[(Int, Int)] =>
+      // Only run this when we have at least 2 items
+      if (items.size > 1) {
+        val keyHits = getHits(Capacity(0), items)
+        assert(keyHits.filter{ _ != 0 }.size == 0)
+      }
+    }
+  }
+
+  property("hit rates in general should be between [0, 1] ") {
+    forAll { (c: Capacity, items: List[(Int, Int)]) =>
+      // Only run this when we have at least 2 items
+      if (items.size > 1) {
+        val keyHits = getHits(c, items)
+        val hitRate = keyHits.sum / items.size.toDouble
+        assert(hitRate >= 0 && hitRate <= 1)
+      }
+    }
+  }
+}
+
 class SummingQueueTest extends PropSpec with PropertyChecks with Matchers {
   val zeroCapQueue = SummingQueue[Int](0) // passes all through