Enforce invariant: mapGroup iterators all nonempty

Previously not all fixes in mapGroup were covered
Not sure I follow...  This refactor makes it so the invariant on
mapGroupNonEmptyIters is that fn(_, Empty) == Empty always

On Mon, Oct 6, 2014 at 12:30 PM, P. Oscar Boykin <notifications@github.com>
wrote:

> In scalding-core/src/main/scala/com/twitter/scalding/typed/Grouped.scala:
>
> >      // don't make a closure
> >      val localRed = reduceFn
> > -    val newReduce = { (k: K, iter: Iterator[V1]) =>
> > -      val step1 = localRed(k, iter)
> > -      // Only pass non-Empty iterators to subsequent functions
> > -      if (step1.nonEmpty) fn(k, step1) else Iterator.empty
> > -    }
> > +    val newReduce = { (k: K, iter: Iterator[V1]) => fn(k, localRed(k, iter)) }
>
> I don't think this works actually. Consider the case where localRed(k,
> iter) == Iterator.empty, but fn(k, _) == Iterator(k). The newReduce will
> not behave as expected. I think your previous patch is correct: If any
> function maps to Iterator.empty, that is the end of the line.
>
> --
> Reply to this email directly or view it on GitHub
> <https://github.com/twitter/scalding/pull/1072/files#r18467483>.
>