re-add the unpackBijectedMonoid logic

[sritchie]

Looks like we forgot this. This should make the generated scalding job quicker:

https://github.com/twitter/summingbird/blob/master/summingbird-core/src/main/scala/com/twitter/summingbird/scalding/BatchAggregatorJob.scala#L101

[sritchie]

Relevant code:

/**
    * Performs a "sum" on the grouped pipe and filters out all
    * zeros. If the supplied Monoid is an instance of
    * BijectedMonoid[U, V], the method will extract the underlying
    * Monoid[U], convert all values over to U, run the monoid and
    * convert back before returning the TypedPipe. This should help
    * with latency in almost all cases that use a BijectedMonoid.
    */
  def sum[U](pipe: Grouped[K, V]): TypedPipe[(K, V)] =
    unpackBijectedMonoid[U].map { case (bijection, monoid) =>
      implicit val m: Monoid[U] = monoid
      pipe.mapValues(bijection.invert(_))
        .sum
        .filter { case (_, u) => Monoid.isNonZero(u) }
        .map { case (k, u) => k -> bijection(u) }
    }.getOrElse {
      pipe.sum // over the V.
        .filter { case (k, v) => Monoid.isNonZero(v) }
    }

  /**
    * If the supplied implicit Monoid[V] is an instance of
    * BijectedMonoid[U, V], Returns the backing Bijection[U, V] and
    * Monoid[U]; else None.
    *
    * TODO: Make the referenced fields on
    * com.twitter.bijection.algebird.Bijected{Monoid,Semigroup,Ring,Group}
    * into vals and remove the reflection here.
    */
  def unpackBijectedMonoid[U]: Option[(Bijection[U, V], Monoid[U])] = {
    implicitly[Monoid[V]] match {
      case m: BijectedMonoid[_, _] => {
        def getField[F: ClassManifest](fieldName: String): F = {
          val f = classOf[BijectedMonoid[_, _]].getDeclaredField(fieldName)
          f.setAccessible(true)
          implicitly[ClassManifest[F]]
            .erasure.asInstanceOf[Class[F]].cast(f.get(m))
        }
        Some((
          getField[ImplicitBijection[U, V]]("bij").bijection,
          getField[Monoid[U]]("monoid")
        ))
      }
      case _ => None
    }
  }

[johnynek]

We put code in algebird to note need these reflection hacks:

https://github.com/twitter/algebird/blob/develop/algebird-bijection/src/main/scala/com/twitter/algebird/bijection/AlgebirdBijections.scala#L32

[sritchie]

For sure... still good to case match, and we can do the same thing within the summingqueue.